Fix SQL injection in connector sample path and validate source queries

amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftConnector.java

@@ -82,7 +82,7 @@ protected SchemaReader getSchemaReader(String sessionID) {
 
   @Override
   protected String getTableName(String database, String schema, String table) {
-    return String.format("\"%s\".\"%s\"", schema, table);
+    return quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"');
   }
 
   @Override

amazon-redshift-plugin/src/test/java/io/cdap/plugin/amazon/redshift/RedshiftConnectorUnitTest.java

@@ -41,6 +41,15 @@ public void getTableNameTest() {
                         CONNECTOR.getTableName("db", "schema", "table"));
   }
 
+  /**
+   * Unit test for getTableName() with identifiers containing double quotes
+   */
+  @Test
+  public void getTableNameEscapesQuotesTest() {
+    Assert.assertEquals("\"my\"\"schema\".\"t\"\"; DELETE FROM x; --\"",
+                        CONNECTOR.getTableName("db", "my\"schema", "t\"; DELETE FROM x; --"));
+  }
+
   @Test
   public void getRandomQuery() {
     Assert.assertEquals("SELECT * FROM TestData\n" +

aurora-mysql-plugin/src/main/java/io/cdap/plugin/auroradb/mysql/AuroraMysqlSource.java

@@ -49,6 +49,12 @@ protected String createConnectionString() {
                          auroraMysqlSourceConfig.host, auroraMysqlSourceConfig.port, auroraMysqlSourceConfig.database);
   }
 
+  @Override
+  protected Boolean useBackslashEscapes() {
+    // In MySQL/MariaDB/SingleStore/Databricks a backslash escapes a quote inside a string literal.
+    return Boolean.TRUE;
+  }
+
   /**
    * Aurora DB MySQL source config.
    */

cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties

@@ -18,8 +18,7 @@ validationErrorMessage=COUNT ERROR found
 errorLogsMessageInvalidTableName=Spark program 'phase-1' failed with error: Stage 'CloudSQL MySQL' encountered : io.cdap.cdap.etl.api.validation.ValidationException: Errors were encountered during validation. \
   Table 'Table123' does not exist.. Please check the system logs for more details.
 errorLogsMessageInvalidCredentials=Spark program 'phase-1' failed with error: Stage 'CloudSQL MySQL' encountered : io.cdap.cdap.api.exception.ProgramFailureException: SQL Error occurred, sqlState: '28000', errorCode: '1045', errorMessage: SQL Exception occurred: [Message='Access denied for user 'testUser'
-errorLogsMessageInvalidBoundingQuery=Spark program 'phase-1' failed with error: Stage 'CloudSQL MySQL' encountered : java.io.IOException: You have an error in your SQL syntax; \
-  check the manual that corresponds to your MySQL server version for the right syntax to use near 'table' at line 1. Please check the system logs for more details.
+errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid Bounding Query
 errorMessageInvalidPassword=SQL Error occurred, sqlState: '28000', errorCode: '1045', errorMessage: SQL Exception occurred: [Message='Access denied for user '
 errorMessagePrivateConnectionName=Enter the internal IP address of the Compute Engine VM cloudsql proxy is running on, to connect to a private 
 errorMessageWithBlankPassword=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '28000', errorCode: '1045', errorMessage: SQL Exception occurred: [Message='Access denied for user '

cloudsql-mysql-plugin/src/e2e-test/resources/pluginParameters.properties

@@ -34,7 +34,7 @@ invalidTable=data
 bqTruncateTable=truncateTable
 bqUpdateTableSchema=updateSchema
 invalidDatabaseName=invalidDB%$^%*
-invalidboundQuery=SELECT MIN(id),MAX(id) FROM table
+invalidboundQuery=SELECT MIN(id), MAX(id) FROM mytable; DELETE FROM mytable
 cloudsqlimportQuery=where $CONDITIONS;
 splitby=ID
 numbersplitsgenerate=2

cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java

@@ -82,7 +82,7 @@ protected SchemaReader getSchemaReader(String sessionID) {
 
   @Override
   protected String getTableName(String database, String schema, String table) {
-    return String.format("`%s`.`%s`", database, table);
+    return quoteIdentifier(database, '`') + "." + quoteIdentifier(table, '`');
   }
 
   @Override

cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLSource.java

@@ -83,6 +83,12 @@ protected Class<? extends DBWritable> getDBRecordType() {
     return MysqlDBRecord.class;
   }
 
+  @Override
+  protected Boolean useBackslashEscapes() {
+    // In MySQL a backslash escapes a quote inside a string literal.
+    return Boolean.TRUE;
+  }
+
   @Override
   protected String createConnectionString() {
     if (CloudSQLUtil.PRIVATE_INSTANCE.equalsIgnoreCase(

cloudsql-mysql-plugin/src/test/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorUnitTest.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.cloudsql.mysql;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link CloudSQLMySQLConnector}
+ */
+public class CloudSQLMySQLConnectorUnitTest {
+
+  private static final CloudSQLMySQLConnector CONNECTOR = new CloudSQLMySQLConnector(
+    new CloudSQLMySQLConnectorConfig("user", "password", "cloudsql-mysql", "", "public",
+                                     "project:region:instance", "db", null));
+
+  /**
+   * Unit test for getTableName()
+   */
+  @Test
+  public void getTableNameTest() {
+    Assert.assertEquals("`db`.`table`",
+                        CONNECTOR.getTableName("db", null, "table"));
+  }
+
+  /**
+   * Unit test for getTableName() with identifiers containing backticks
+   */
+  @Test
+  public void getTableNameEscapesBackticksTest() {
+    Assert.assertEquals("`my``db`.`t``; DELETE FROM x; --`",
+                        CONNECTOR.getTableName("my`db", null, "t`; DELETE FROM x; --"));
+  }
+}

cloudsql-postgresql-plugin/src/e2e-test/resources/errorMessage.properties

@@ -16,7 +16,7 @@ errorMessageInvalidPassword=SQL Error occurred, sqlState: '28P01', errorCode: '0
 errorMessageInvalidSourceHost=SQL error while getting query schema: The connection attempt failed.
 errorMessageInvalidTableName=Table 'table' does not exist. Ensure table '"table"' is set correctly and that the
 errorMessageInvalidSinkDatabase=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '3D000', errorCode: '0', errorMessage: SQL Exception occurred: [Message='FATAL: database "invalidDB" does not exist', SQLState='3D000', ErrorCode='0'].'
-errorLogsMessageInvalidBoundingQuery=The column index is out of range: 1, number of columns: 0.. 
+errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid Bounding Query
 errorMessageConnectionName=Connection Name must be in the format <PROJECT_ID>:<REGION>:<INSTANCE_NAME> to connect to \
   a public CloudSQL PostgreSQL instance.
 errorMessagePrivateConnectionName=Enter the internal IP address of the Compute Engine VM cloudsql proxy is running on, \

cloudsql-postgresql-plugin/src/e2e-test/resources/pluginParameters.properties

@@ -68,7 +68,7 @@ invalidDriverName=abcdriver
 invalidUserName=testUser
 invalidPassword=testPassword
 invalidBoundingQuery=SELECT MIN(id),MAX(id) FROM table
-invalidBoundingQueryValue=select;
+invalidBoundingQueryValue=SELECT MIN(id), MAX(id) FROM mytable; DELETE FROM mytable
 invalidTable=table
 invalidConnectionName=abd3
 invalidCloudSQLPostgreSQLTableKey=ss

cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java

@@ -86,7 +86,7 @@ protected SchemaReader getSchemaReader(String sessionID) {
 
   @Override
   protected String getTableName(String database, String schema, String table) {
-    return String.format("\"%s\".\"%s\"", schema, table);
+    return quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"');
   }
 
   @Override

cloudsql-postgresql-plugin/src/test/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnectorUnitTest.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.cloudsql.postgres;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link CloudSQLPostgreSQLConnector}
+ */
+public class CloudSQLPostgreSQLConnectorUnitTest {
+
+  private static final CloudSQLPostgreSQLConnector CONNECTOR = new CloudSQLPostgreSQLConnector(
+    new CloudSQLPostgreSQLConnectorConfig("user", "password", "cloudsql-postgresql", "", "public",
+                                          "project:region:instance", "db", null));
+
+  /**
+   * Unit test for getTableName()
+   */
+  @Test
+  public void getTableNameTest() {
+    Assert.assertEquals("\"schema\".\"table\"",
+                        CONNECTOR.getTableName("db", "schema", "table"));
+  }
+
+  /**
+   * Unit test for getTableName() with identifiers containing double quotes, using the payload from the
+   * reported sample path SQL injection
+   */
+  @Test
+  public void getTableNameEscapesQuotesTest() {
+    String table = "pg_class\" LIMIT 0; UPDATE audit_vault.write_marker SET value='INJECTED_BY_SAMPLE_PATH' " +
+      "WHERE id=1; SELECT * FROM \"pg_catalog\".\"pg_class";
+    Assert.assertEquals("\"pg_catalog\".\"pg_class\"\" LIMIT 0; UPDATE audit_vault.write_marker SET " +
+                          "value='INJECTED_BY_SAMPLE_PATH' WHERE id=1; SELECT * FROM \"\"pg_catalog\"\".\"\"pg_class\"",
+                        CONNECTOR.getTableName("db", "pg_catalog", table));
+    Assert.assertEquals("\"my\"\"schema\".\"table\"",
+                        CONNECTOR.getTableName("db", "my\"schema", "table"));
+  }
+}

database-commons/pom.xml

@@ -77,6 +77,10 @@
       <groupId>dev.failsafe</groupId>
       <artifactId>failsafe</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.github.jsqlparser</groupId>
+      <artifactId>jsqlparser</artifactId>
+    </dependency>
 
     <!-- test dependencies -->
     <dependency>

database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java

@@ -48,6 +48,7 @@
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.Map;
+import java.util.Objects;
 import java.util.UUID;
 import javax.annotation.Nullable;
 
@@ -150,8 +151,18 @@ protected String getConnectionString(String database) {
   }
 
   protected String getTableName(String database, String schema, String table) {
-    return schema == null ? String.format("\"%s\".\"%s\"", database, table)
-      : String.format("\"%s\".\"%s\".\"%s\"", database, schema, table);
+    return schema == null ? quoteIdentifier(database, '"') + "." + quoteIdentifier(table, '"')
+      : quoteIdentifier(database, '"') + "." + quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"');
+  }
+
+  /**
+   * Wraps the given identifier in the given quote character, doubling any embedded quote characters so that a
+   * user supplied database, schema, table or column name can never break out of the quoted identifier.
+   */
+  protected static String quoteIdentifier(String identifier, char quote) {
+    Objects.requireNonNull(identifier, "identifier to quote must not be null");
+    String quoteString = String.valueOf(quote);
+    return quoteString + identifier.replace(quoteString, quoteString + quoteString) + quoteString;
   }
 
   protected String getTableQuery(String database, String schema, String table) {

database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java

@@ -52,6 +52,7 @@
 import io.cdap.plugin.util.DriverCleanup;
 import io.cdap.plugin.util.RetryPolicyUtil;
 import io.cdap.plugin.util.RetryUtils;
+import io.cdap.plugin.util.SelectQueryValidator;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
@@ -106,6 +107,10 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
     StageConfigurer stageConfigurer = pipelineConfigurer.getStageConfigurer();
     FailureCollector collector = stageConfigurer.getFailureCollector();
     sourceConfig.validate(collector);
+    // The import query is executed at configure time (schema fetch) and the split-by is embedded in generated SQL,
+    // so both are validated here. The bounding query is only used at runtime and is validated in prepareRun.
+    validateQuery(DBSourceConfig.IMPORT_QUERY, "Import Query", sourceConfig.getImportQuery(), collector);
+    validateSplitBy(collector);
     if (sourceConfig.getSchema() != null) {
       stageConfigurer.setOutputSchema(sourceConfig.getSchema());
       return;
@@ -119,7 +124,8 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
     Class<? extends Driver> driverClass = DBUtils.getDriverClass(
       pipelineConfigurer, sourceConfig, ConnectionConfig.JDBC_PLUGIN_TYPE);
 
-    if (sourceConfig.canConnect()) {
+    // Do not open a connection or run init queries if the import/bounding/split-by validation already failed.
+    if (sourceConfig.canConnect() && collector.getValidationFailures().isEmpty()) {
       try {
         stageConfigurer.setOutputSchema(getSchema(driverClass));
       } catch (IllegalAccessException | InstantiationException e) {
@@ -161,18 +167,32 @@ public Schema getSchema(Class<? extends Driver> driverClass) throws IllegalAcces
    *                      3. Error while running the import query.
    */
   public Schema getSchema() throws SQLException {
+    // Validate the import query before opening a connection or running init queries, so a non-SELECT query never
+    // reaches the database and init queries do not run for a query that will be rejected.
+    validateImportQuery(sourceConfig.getImportQuery());
     try (Connection connection = getConnection()) {
       executeInitQueries(connection, sourceConfig.getInitQueries());
       String query = sourceConfig.getImportQuery();
       return loadSchemaFromDB(connection, query);
     }
   }
 
+  private void validateImportQuery(String query) {
+    String finalQuery = query != null && query.contains("$CONDITIONS") ? removeConditionsClause(query) : query;
+    try {
+      SelectQueryValidator.validate(finalQuery, useSquareBracketQuotes(), useBackslashEscapes());
+    } catch (IllegalArgumentException e) {
+      throw new IllegalArgumentException("Invalid Import Query: " + e.getMessage(), e);
+    }
+  }
+
   private Schema loadSchemaFromDB(Connection connection, String query) throws SQLException {
+    String finalQuery = query.contains("$CONDITIONS") ? removeConditionsClause(query) : query;
+    // never send a query that is not a single read-only SELECT to the database
+    validateImportQuery(query);
     try (Statement statement = RetryUtils.createStatementWithRetry((RetryPolicy<Statement>) retryPolicy, connection,
       getErrorDetailsProvider())) {
       statement.setMaxRows(1);
-      String finalQuery = query.contains("$CONDITIONS") ? removeConditionsClause(query) : query;
       ResultSet resultSet = RetryUtils.executeQueryWithRetry((RetryPolicy<ResultSet>) retryPolicy, statement,
         finalQuery, getErrorDetailsProvider());
       return Schema.recordOf("outputSchema", getSchemaReader().getSchemaFields(resultSet));
@@ -228,6 +248,72 @@ protected String getErrorDetailsProviderClassName() {
     return null;
   }
 
+  /**
+   * Returns whether square brackets quote identifiers in the database's SQL dialect, as in SQL Server.
+   * Override this method to allow such identifiers in the import and bounding queries.
+   */
+  protected boolean useSquareBracketQuotes() {
+    return false;
+  }
+
+  /**
+   * Validates that the import and bounding queries are single read-only SELECT statements before the pipeline runs,
+   * since they are executed with the credential of the connection.
+   */
+  private void validateQueries(FailureCollector collector) {
+    validateQuery(DBSourceConfig.IMPORT_QUERY, "Import Query", sourceConfig.getImportQuery(), collector);
+    validateQuery(DBSourceConfig.BOUNDING_QUERY, "Bounding Query", sourceConfig.getBoundingQuery(), collector);
+    validateSplitBy(collector);
+  }
+
+  private void validateQuery(String property, String label, @Nullable String query, FailureCollector collector) {
+    // A macro is not resolved at configure time; it is validated at runtime once it holds a concrete value.
+    if (sourceConfig.containsMacro(property) || Strings.isNullOrEmpty(query)) {
+      return;
+    }
+    try {
+      SelectQueryValidator.validate(query, useSquareBracketQuotes(), useBackslashEscapes());
+    } catch (IllegalArgumentException e) {
+      collector.addFailure(String.format("Invalid %s: %s", label, e.getMessage()),
+                           "Provide a single SELECT statement that only reads data.")
+        .withConfigProperty(property);
+    }
+  }
+
+  /**
+   * Validates that the split-by field is a single column name. When numSplits is not 1, the split-by value is
+   * concatenated into the generated split bound clauses at runtime (via {@code $CONDITIONS} substitution) and
+   * executed with the connection credential, so it must not be able to carry extra SQL.
+   */
+  private void validateSplitBy(FailureCollector collector) {
+    if (sourceConfig.containsMacro(DBSourceConfig.SPLIT_BY) || sourceConfig.containsMacro(DBSourceConfig.NUM_SPLITS)) {
+      return;
+    }
+    Integer numSplits = sourceConfig.getNumSplits();
+    String splitBy = sourceConfig.getSplitBy();
+    if ((numSplits != null && numSplits == 1) || Strings.isNullOrEmpty(splitBy)) {
+      return;
+    }
+    try {
+      SelectQueryValidator.validateColumnReference(splitBy, "Split-By Field Name", useSquareBracketQuotes());
+    } catch (IllegalArgumentException e) {
+      collector.addFailure(String.format("Invalid Split-By Field Name: %s", e.getMessage()),
+                           "Provide a single column name.")
+        .withConfigProperty(DBSourceConfig.SPLIT_BY);
+    }
+  }
+
+  /**
+   * Returns whether a backslash escapes a quote inside a string literal in the database's dialect, so that source
+   * queries are parsed the way the database driver parses them. Returns {@code null} when the dialect is unknown,
+   * in which case the query must be valid under both interpretations. MySQL/MariaDB/SingleStore/Databricks sources
+   * override this to return {@code true}.
+   */
+  @Nullable
+  protected Boolean useBackslashEscapes() {
+    return null;
+  }
+
   /**
    * Returns the DBErrorDetailsProvider instance.
    * Override this method to provide a custom DBErrorDetailsProvider instance.
@@ -272,6 +358,7 @@ private Connection getConnection() {
   public void prepareRun(BatchSourceContext context) throws Exception {
     FailureCollector collector = context.getFailureCollector();
     sourceConfig.validate(collector);
+    validateQueries(collector);
     collector.getOrThrowException();
 
     LOG.debug("pluginType = {}; pluginName = {}; connectionString = {}; importQuery = {}; " +