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" +

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
 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/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/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/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

@@ -150,8 +150,17 @@ 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.
+   */
+  public static String quoteIdentifier(String identifier, char quote) {
+    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;
@@ -169,10 +170,16 @@ public Schema getSchema() throws SQLException {
   }
 
   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
+    try {
+      SelectQueryValidator.validate(finalQuery, useSquareBracketQuotes());
+    } catch (IllegalArgumentException e) {
+      throw new IllegalArgumentException("Invalid Import Query: " + e.getMessage(), e);
+    }
     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 +235,36 @@ 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);
+  }
+
+  private void validateQuery(String property, String label, @Nullable String query, FailureCollector collector) {
+    if (Strings.isNullOrEmpty(query)) {
+      return;
+    }
+    try {
+      SelectQueryValidator.validate(query, useSquareBracketQuotes());
+    } catch (IllegalArgumentException e) {
+      collector.addFailure(String.format("Invalid %s: %s", label, e.getMessage()),
+                           "Provide a single SELECT statement that only reads data.")
+        .withConfigProperty(property);
+    }
+  }
+
   /**
    * Returns the DBErrorDetailsProvider instance.
    * Override this method to provide a custom DBErrorDetailsProvider instance.
@@ -272,6 +309,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 = {}; " +

database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java

@@ -0,0 +1,133 @@
+/*
+ * 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.util;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import net.sf.jsqlparser.JSQLParserException;
+import net.sf.jsqlparser.parser.CCJSqlParserUtil;
+import net.sf.jsqlparser.statement.Statement;
+import net.sf.jsqlparser.statement.Statements;
+import net.sf.jsqlparser.statement.select.ParenthesedSelect;
+import net.sf.jsqlparser.statement.select.PlainSelect;
+import net.sf.jsqlparser.statement.select.Select;
+import net.sf.jsqlparser.statement.select.SetOperationList;
+import net.sf.jsqlparser.statement.select.WithItem;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.stream.Collectors;
+
+/**
+ * Validates that a user provided query is a single, read-only SELECT statement before it is executed with the
+ * credential of a database connection.
+ */
+public final class SelectQueryValidator {
+
+  private static final long PARSE_TIMEOUT_MS = 10_000L;
+  // Placeholder replaced with the split condition at runtime, it is not valid SQL on its own
+  private static final String CONDITIONS_PLACEHOLDER = "$CONDITIONS";
+  private static final ThreadFactory PARSER_THREAD_FACTORY = new ThreadFactoryBuilder()
+    .setDaemon(true).setNameFormat("select-query-validator-%d").build();
+
+  private SelectQueryValidator() {
+  }
+
+  /**
+   * Validates that the given query is a single read-only SELECT statement.
+   *
+   * @param query the query to validate
+   * @param squareBracketQuotes whether square brackets quote identifiers, as in SQL Server
+   * @throws IllegalArgumentException if the query cannot be parsed or is not a single read-only SELECT statement
+   */
+  public static void validate(String query, boolean squareBracketQuotes) {
+    String parsableQuery = query.replace(CONDITIONS_PLACEHOLDER, "1 = 1");
+    // Databases disagree on whether a backslash escapes a quote inside a string literal. Requiring a single SELECT
+    // under both interpretations makes sure a crafted string literal cannot hide a second statement.
+    validate(parsableQuery, squareBracketQuotes, false);
+    validate(parsableQuery, squareBracketQuotes, true);
+  }
+
+  private static void validate(String query, boolean squareBracketQuotes, boolean backslashEscapes) {
+    Statements statements;
+    // Use our own executor so that its thread is always released, the parser does not shut down its own executor
+    // when parsing fails.
+    ExecutorService executor = Executors.newSingleThreadExecutor(PARSER_THREAD_FACTORY);
+    try {
+      statements = CCJSqlParserUtil.parseStatements(query, executor, parser -> parser
+        .withSquareBracketQuotation(squareBracketQuotes)
+        .withBackslashEscapeCharacter(backslashEscapes)
+        .withTimeOut(PARSE_TIMEOUT_MS));
+    } catch (JSQLParserException e) {
+      throw new IllegalArgumentException(
+        "Query could not be parsed. Only a single SELECT statement is allowed. Error: " + firstLine(e), e);
+    } finally {
+      executor.shutdownNow();
+    }
+
+    // the parser returns null for an empty query
+    List<Statement> nonEmptyStatements = statements == null ? Collections.emptyList()
+      : statements.stream().filter(Objects::nonNull).collect(Collectors.toList());
+    if (nonEmptyStatements.size() != 1) {
+      throw new IllegalArgumentException(String.format(
+        "Only a single SELECT statement is allowed, but found %d statements.", nonEmptyStatements.size()));
+    }
+    Statement statement = nonEmptyStatements.get(0);
+    if (!(statement instanceof Select)) {
+      throw new IllegalArgumentException(String.format(
+        "Only a SELECT statement is allowed, but found a %s statement.", statement.getClass().getSimpleName()));
+    }
+    validateSelect((Select) statement);
+  }
+
+  private static void validateSelect(Select select) {
+    if (select.getWithItemsList() != null) {
+      for (WithItem withItem : select.getWithItemsList()) {
+        validateSelect(withItem.getSelect());
+      }
+    }
+    if (select instanceof PlainSelect) {
+      PlainSelect plainSelect = (PlainSelect) select;
+      if ((plainSelect.getIntoTables() != null && !plainSelect.getIntoTables().isEmpty())
+        || plainSelect.getIntoTempTable() != null) {
+        throw new IllegalArgumentException("SELECT ... INTO is not allowed, the query must only read data.");
+      }
+      if (plainSelect.getForMode() != null) {
+        throw new IllegalArgumentException(String.format(
+          "SELECT ... FOR %s is not allowed, the query must only read data.", plainSelect.getForMode().getValue()));
+      }
+    } else if (select instanceof SetOperationList) {
+      for (Select operand : ((SetOperationList) select).getSelects()) {
+        validateSelect(operand);
+      }
+    } else if (select instanceof ParenthesedSelect) {
+      validateSelect(((ParenthesedSelect) select).getSelect());
+    } else {
+      throw new IllegalArgumentException(String.format(
+        "Unsupported SELECT statement of type %s.", select.getClass().getSimpleName()));
+    }
+  }
+
+  private static String firstLine(Exception e) {
+    String message = String.valueOf(e.getMessage());
+    int newLine = message.indexOf('\n');
+    return newLine < 0 ? message : message.substring(0, newLine);
+  }
+}