diff --git a/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/SecretsRepositoryWriter.java b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/SecretsRepositoryWriter.java
index afe718dbba1b..d0d0fa4f5ee9 100644
--- a/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/SecretsRepositoryWriter.java
+++ b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/SecretsRepositoryWriter.java
@@ -162,13 +162,13 @@ private JsonNode statefulUpdateSecrets(final UUID workspaceId,
           workspaceId,
           oldConfig.get(),
           fullConfig,
-          spec,
+          spec.getConnectionSpecification(),
           longLivedSecretPersistence.get());
     } else {
       splitSecretConfig = SecretsHelpers.splitConfig(
           workspaceId,
           fullConfig,
-          spec);
+          spec.getConnectionSpecification());
     }
     splitSecretConfig.getCoordinateToPayload().forEach(longLivedSecretPersistence.get()::write);
     return splitSecretConfig.getPartialConfig();
@@ -188,7 +188,7 @@ private JsonNode splitSecretConfig(final UUID workspaceId,
                                      final ConnectorSpecification spec,
                                      final Optional<SecretPersistence> secretPersistence) {
     if (secretPersistence.isPresent()) {
-      final SplitSecretConfig splitSecretConfig = SecretsHelpers.splitConfig(workspaceId, fullConfig, spec);
+      final SplitSecretConfig splitSecretConfig = SecretsHelpers.splitConfig(workspaceId, fullConfig, spec.getConnectionSpecification());
       splitSecretConfig.getCoordinateToPayload().forEach(secretPersistence.get()::write);
       return splitSecretConfig.getPartialConfig();
     } else {
diff --git a/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/SecretsHelpers.java b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/SecretsHelpers.java
index 1300bc8d5c8a..c8d975c27df1 100644
--- a/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/SecretsHelpers.java
+++ b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/SecretsHelpers.java
@@ -14,7 +14,6 @@
 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 java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -27,18 +26,17 @@
  * Contains most of the core logic surrounding secret coordinate extraction and insertion.
  *
  * These are the three main helpers provided by this class:
- * {@link SecretsHelpers#splitConfig(UUID, JsonNode, ConnectorSpecification)}
- * {@link SecretsHelpers#splitAndUpdateConfig(UUID, JsonNode, JsonNode, ConnectorSpecification, ReadOnlySecretPersistence)}
+ * {@link SecretsHelpers#splitConfig(UUID, JsonNode, JsonNode)}
+ * {@link SecretsHelpers#splitAndUpdateConfig(UUID, JsonNode, JsonNode, JsonNode, ReadOnlySecretPersistence)}
  * {@link SecretsHelpers#combineConfig(JsonNode, ReadOnlySecretPersistence)}
  *
  * Here's an overview on some terminology used in this class:
  *
- * A "full config" represents an entire connector config as specified by an end user. This should
- * conform to a connector specification.
+ * A "full config" represents an entire config as specified by an end user.
  *
- * A "partial config" represents a connector config where any string that was specified as an
- * airbyte_secret in the connector specification is replaced by a JSON object {"_secret": "secret
- * coordinate"} that can later be used to reconstruct the "full config".
+ * A "partial config" represents a config where any string that was specified as an airbyte_secret
+ * in the specification is replaced by a JSON object {"_secret": "secret coordinate"} that can later
+ * be used to reconstruct the "full config".
  *
  * A {@link SecretPersistence} provides the ability to read and write secrets to a backing store
  * such as Google Secrets Manager.
@@ -52,41 +50,39 @@ public class SecretsHelpers {
   public static final String COORDINATE_FIELD = "_secret";
 
   /**
-   * Used to separate secrets out of a connector configuration. This will output a partial config that
+   * Used to separate secrets out of some configuration. This will output a partial config that
    * includes pointers to secrets instead of actual secret values and a map that can be used to update
    * a {@link SecretPersistence} at coordinates with values from the full config.
    *
-   * @param workspaceId workspace used for this connector config
+   * @param workspaceId workspace used for this config
    * @param fullConfig config including secrets
-   * @param spec connector specification
+   * @param spec specification for the config
    * @return a partial config + a map of coordinates to secret payloads
    */
   public static SplitSecretConfig splitConfig(final UUID workspaceId,
                                               final JsonNode fullConfig,
-                                              final ConnectorSpecification spec) {
+                                              final JsonNode spec) {
     return internalSplitAndUpdateConfig(
         UUID::randomUUID,
         workspaceId,
         (coordinate) -> Optional.empty(),
         Jsons.emptyObject(),
         fullConfig,
-        spec.getConnectionSpecification());
+        spec);
   }
 
   /**
-   * Used to separate secrets out of a connector configuration and output a partial config that
-   * includes pointers to secrets instead of actual secret values and a map that can be used to update
-   * a {@link SecretPersistence} at coordinates with values from the full config. If a previous config
-   * for this connector's configuration is provided, this method attempts to use the same base
-   * coordinates to refer to the same secret and increment the version of the coordinate used to
-   * reference a secret.
+   * Used to separate secrets out of a configuration and output a partial config that includes
+   * pointers to secrets instead of actual secret values and a map that can be used to update a
+   * {@link SecretPersistence} at coordinates with values from the full config. If a previous config
+   * for this configuration is provided, this method attempts to use the same base coordinates to
+   * refer to the same secret and increment the version of the coordinate used to reference a secret.
    *
-   * @param workspaceId workspace used for this connector config
-   * @param oldPartialConfig previous partial config for this specific connector configuration
+   * @param workspaceId workspace used for this config
+   * @param oldPartialConfig previous partial config for this specific configuration
    * @param newFullConfig new config containing secrets that will be used to update the partial config
-   *        for this connector
-   * @param spec connector specification that should match both the previous partial config after
-   *        filling in coordinates and the new full config.
+   * @param spec specification that should match both the previous partial config after filling in
+   *        coordinates and the new full config.
    * @param secretReader provides a way to determine if a secret is the same or updated at a specific
    *        location in a config
    * @return a partial config + a map of coordinates to secret payloads
@@ -94,7 +90,7 @@ public static SplitSecretConfig splitConfig(final UUID workspaceId,
   public static SplitSecretConfig splitAndUpdateConfig(final UUID workspaceId,
                                                        final JsonNode oldPartialConfig,
                                                        final JsonNode newFullConfig,
-                                                       final ConnectorSpecification spec,
+                                                       final JsonNode spec,
                                                        final ReadOnlySecretPersistence secretReader) {
     return internalSplitAndUpdateConfig(
         UUID::randomUUID,
@@ -102,7 +98,7 @@ public static SplitSecretConfig splitAndUpdateConfig(final UUID workspaceId,
         secretReader,
         oldPartialConfig,
         newFullConfig,
-        spec.getConnectionSpecification());
+        spec);
   }
 
   /**
@@ -148,9 +144,9 @@ public static JsonNode combineConfig(final JsonNode partialConfig, final ReadOnl
   public static SplitSecretConfig splitConfig(final Supplier<UUID> uuidSupplier,
                                               final UUID workspaceId,
                                               final JsonNode fullConfig,
-                                              final ConnectorSpecification spec) {
+                                              final JsonNode spec) {
     return internalSplitAndUpdateConfig(uuidSupplier, workspaceId, (coordinate) -> Optional.empty(), Jsons.emptyObject(), fullConfig,
-        spec.getConnectionSpecification());
+        spec);
   }
 
   /**
@@ -162,9 +158,9 @@ public static SplitSecretConfig splitAndUpdateConfig(final Supplier<UUID> uuidSu
                                                        final UUID workspaceId,
                                                        final JsonNode oldPartialConfig,
                                                        final JsonNode newFullConfig,
-                                                       final ConnectorSpecification spec,
+                                                       final JsonNode spec,
                                                        final ReadOnlySecretPersistence secretReader) {
-    return internalSplitAndUpdateConfig(uuidSupplier, workspaceId, secretReader, oldPartialConfig, newFullConfig, spec.getConnectionSpecification());
+    return internalSplitAndUpdateConfig(uuidSupplier, workspaceId, secretReader, oldPartialConfig, newFullConfig, spec);
   }
 
   /**
@@ -207,7 +203,7 @@ private static SecretCoordinate getOrCreateCoordinate(final ReadOnlySecretPersis
    *
    * For splits that don't have a prior partial config (such as when a connector is created for a
    * source or destination for the first time), the secret reader and old partial config can be set to
-   * empty (see {@link SecretsHelpers#splitConfig(UUID, JsonNode, ConnectorSpecification)}).
+   * empty (see {@link SecretsHelpers#splitConfig(UUID, JsonNode, JsonNode)}).
    *
    * IMPORTANT: This is used recursively. In the process, the partial config, full config, and spec
    * inputs will represent internal json structures, not the entire configs/specs.
@@ -218,10 +214,9 @@ private static SecretCoordinate getOrCreateCoordinate(final ReadOnlySecretPersis
    *        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 persistedPartialConfig previous partial config for this specific connector configuration
+   * @param persistedPartialConfig previous partial config for this specific configuration
    * @param newFullConfig new config containing secrets that will be used to update the partial config
-   *        for this connector
-   * @param spec connector specification
+   * @param spec config specification
    * @return a partial config + a map of coordinates to secret payloads
    */
   @VisibleForTesting
@@ -292,7 +287,7 @@ private static SecretCoordinate getCoordinateFromTextNode(final JsonNode node) {
    *
    * @param newSecret new value of a secret provides a way to determine if a secret is the same or
    *        updated at a specific location in a config
-   * @param workspaceId workspace used for this connector config
+   * @param workspaceId workspace used for this config
    * @param uuidSupplier provided to allow a test case to produce known UUIDs in order for easy
    *        fixture creation.
    * @param oldSecretFullCoordinate a nullable full coordinate (base+version) retrieved from the
diff --git a/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsHelpersTest.java b/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsHelpersTest.java
index 4658730bd365..ec2784532233 100644
--- a/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsHelpersTest.java
+++ b/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/split_secrets/SecretsHelpersTest.java
@@ -88,7 +88,7 @@ void testSplit(final SecretsTestCase testCase) {
         uuidIterator::next,
         WORKSPACE_ID,
         inputConfig,
-        testCase.getSpec());
+        testCase.getSpec().getConnectionSpecification());
 
     assertEquals(testCase.getPartialConfig(), splitConfig.getPartialConfig());
     assertEquals(testCase.getFirstSecretMap(), splitConfig.getCoordinateToPayload());
@@ -131,7 +131,7 @@ void testSplitUpdate(final SecretsTestCase testCase) {
         WORKSPACE_ID,
         inputPartialConfig,
         inputUpdateConfig,
-        testCase.getSpec(),
+        testCase.getSpec().getConnectionSpecification(),
         secretPersistence::read);
 
     assertEquals(testCase.getUpdatedPartialConfig(), updatedSplit.getPartialConfig());
@@ -179,7 +179,7 @@ void testUpdatingSecretsOneAtATime() {
         uuidIterator::next,
         WORKSPACE_ID,
         testCase.getFullConfig(),
-        testCase.getSpec());
+        testCase.getSpec().getConnectionSpecification());
 
     assertEquals(testCase.getPartialConfig(), splitConfig.getPartialConfig());
     assertEquals(testCase.getFirstSecretMap(), splitConfig.getCoordinateToPayload());
@@ -193,7 +193,7 @@ void testUpdatingSecretsOneAtATime() {
         WORKSPACE_ID,
         testCase.getPartialConfig(),
         testCase.getFullConfigUpdate1(),
-        testCase.getSpec(),
+        testCase.getSpec().getConnectionSpecification(),
         secretPersistence::read);
 
     assertEquals(testCase.getUpdatedPartialConfigAfterUpdate1(), updatedSplit1.getPartialConfig());
@@ -208,7 +208,7 @@ void testUpdatingSecretsOneAtATime() {
         WORKSPACE_ID,
         updatedSplit1.getPartialConfig(),
         testCase.getFullConfigUpdate2(),
-        testCase.getSpec(),
+        testCase.getSpec().getConnectionSpecification(),
         secretPersistence::read);
 
     assertEquals(testCase.getUpdatedPartialConfigAfterUpdate2(), updatedSplit2.getPartialConfig());