From 8e7560eaede2807cfd9d8d659c7bf3bdd5935557 Mon Sep 17 00:00:00 2001 From: vishwasvaidya-cloudsufi Date: Fri, 30 Oct 2026 14:11:07 +0530 Subject: [PATCH 1/2] Fix SQL injection in connector sample path and validate source queries --- .../amazon/redshift/RedshiftConnector.java | 2 +- .../redshift/RedshiftConnectorUnitTest.java | 9 ++ .../resources/errorMessage.properties | 3 +- .../mysql/CloudSQLMySQLConnector.java | 2 +- .../mysql/CloudSQLMySQLConnectorUnitTest.java | 48 +++++++ .../postgres/CloudSQLPostgreSQLConnector.java | 2 +- .../CloudSQLPostgreSQLConnectorUnitTest.java | 54 +++++++ database-commons/pom.xml | 4 + .../AbstractDBSpecificConnector.java | 13 +- .../plugin/db/source/AbstractDBSource.java | 40 +++++- .../plugin/util/SelectQueryValidator.java | 133 +++++++++++++++++ .../db/connector/QuoteIdentifierTest.java | 52 +++++++ .../plugin/util/SelectQueryValidatorTest.java | 136 ++++++++++++++++++ .../databricks/DatabricksConnector.java | 13 +- .../DatabricksConnectorUnitTest.java | 19 ++- .../io/cdap/plugin/jdbc/DatabaseSource.java | 13 ++ .../cdap/plugin/db/sink/DBSourceTestRun.java | 63 ++++++++ .../cdap/plugin/jdbc/DatabaseSourceTest.java | 42 ++++++ .../resources/errorMessage.properties | 2 +- .../cdap/plugin/mssql/SqlServerConnector.java | 8 +- .../io/cdap/plugin/mssql/SqlServerSource.java | 5 + .../mssql/SqlServerConnectorUnitTest.java | 21 +++ .../io/cdap/plugin/mysql/MysqlConnector.java | 7 +- .../plugin/mysql/MysqlConnectorUnitTest.java | 19 +++ .../cdap/plugin/oracle/OracleConnector.java | 7 +- .../oracle/OracleConnectorUnitTest.java | 19 +++ pom.xml | 6 + .../resources/errorMessage.properties | 3 +- .../plugin/postgres/PostgresConnector.java | 7 +- .../postgres/PostgresConnectorUnitTest.java | 22 +++ 30 files changed, 743 insertions(+), 31 deletions(-) create mode 100644 cloudsql-mysql-plugin/src/test/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorUnitTest.java create mode 100644 cloudsql-postgresql-plugin/src/test/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnectorUnitTest.java create mode 100644 database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java create mode 100644 database-commons/src/test/java/io/cdap/plugin/db/connector/QuoteIdentifierTest.java create mode 100644 database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java create mode 100644 generic-database-plugin/src/test/java/io/cdap/plugin/jdbc/DatabaseSourceTest.java diff --git a/amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftConnector.java b/amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftConnector.java index fb8cac4a7..57a253ebb 100644 --- a/amazon-redshift-plugin/src/main/java/io/cdap/plugin/amazon/redshift/RedshiftConnector.java +++ b/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 diff --git a/amazon-redshift-plugin/src/test/java/io/cdap/plugin/amazon/redshift/RedshiftConnectorUnitTest.java b/amazon-redshift-plugin/src/test/java/io/cdap/plugin/amazon/redshift/RedshiftConnectorUnitTest.java index 94ea6004b..16c94a3f0 100644 --- a/amazon-redshift-plugin/src/test/java/io/cdap/plugin/amazon/redshift/RedshiftConnectorUnitTest.java +++ b/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" + diff --git a/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties b/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties index ec101ed02..f6ed2205e 100644 --- a/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties +++ b/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 ' diff --git a/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java b/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java index 0751bd160..dccb8b047 100644 --- a/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java +++ b/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 diff --git a/cloudsql-mysql-plugin/src/test/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorUnitTest.java b/cloudsql-mysql-plugin/src/test/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorUnitTest.java new file mode 100644 index 000000000..84284ceb4 --- /dev/null +++ b/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; --")); + } +} diff --git a/cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java b/cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java index 348a1b94c..8f1921478 100644 --- a/cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java +++ b/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 diff --git a/cloudsql-postgresql-plugin/src/test/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnectorUnitTest.java b/cloudsql-postgresql-plugin/src/test/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnectorUnitTest.java new file mode 100644 index 000000000..fa8cb8b6a --- /dev/null +++ b/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")); + } +} diff --git a/database-commons/pom.xml b/database-commons/pom.xml index 98bcb3803..2d92db70d 100644 --- a/database-commons/pom.xml +++ b/database-commons/pom.xml @@ -77,6 +77,10 @@ dev.failsafe failsafe + + com.github.jsqlparser + jsqlparser + diff --git a/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java b/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java index e0407203b..f1b0fdf6f 100644 --- a/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java +++ b/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) { diff --git a/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java b/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java index 908d27eb9..9517bb2b9 100644 --- a/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java +++ b/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) retryPolicy, connection, getErrorDetailsProvider())) { statement.setMaxRows(1); - String finalQuery = query.contains("$CONDITIONS") ? removeConditionsClause(query) : query; ResultSet resultSet = RetryUtils.executeQueryWithRetry((RetryPolicy) 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 = {}; " + diff --git a/database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java b/database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java new file mode 100644 index 000000000..9305d84a1 --- /dev/null +++ b/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 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); + } +} diff --git a/database-commons/src/test/java/io/cdap/plugin/db/connector/QuoteIdentifierTest.java b/database-commons/src/test/java/io/cdap/plugin/db/connector/QuoteIdentifierTest.java new file mode 100644 index 000000000..f92c59c2e --- /dev/null +++ b/database-commons/src/test/java/io/cdap/plugin/db/connector/QuoteIdentifierTest.java @@ -0,0 +1,52 @@ +/* + * 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.db.connector; + +import org.junit.Assert; +import org.junit.Test; + +/** + * Unit tests for {@link AbstractDBSpecificConnector#quoteIdentifier(String, char)} + */ +public class QuoteIdentifierTest { + + @Test + public void testPlainIdentifier() { + Assert.assertEquals("\"table\"", AbstractDBSpecificConnector.quoteIdentifier("table", '"')); + Assert.assertEquals("`table`", AbstractDBSpecificConnector.quoteIdentifier("table", '`')); + } + + @Test + public void testEmbeddedQuotesAreDoubled() { + Assert.assertEquals("\"my\"\"table\"", AbstractDBSpecificConnector.quoteIdentifier("my\"table", '"')); + Assert.assertEquals("`my``table`", AbstractDBSpecificConnector.quoteIdentifier("my`table", '`')); + Assert.assertEquals("\"\"\"\"\"\"", AbstractDBSpecificConnector.quoteIdentifier("\"\"", '"')); + } + + @Test + public void testOtherQuoteCharacterIsLeftAsIs() { + Assert.assertEquals("\"my`table\"", AbstractDBSpecificConnector.quoteIdentifier("my`table", '"')); + Assert.assertEquals("`my\"table`", AbstractDBSpecificConnector.quoteIdentifier("my\"table", '`')); + } + + @Test + public void testInjectionPayloadStaysInsideIdentifier() { + String payload = "pg_class\" LIMIT 0; UPDATE t SET v='x' WHERE id=1; SELECT * FROM \"pg_class"; + Assert.assertEquals("\"pg_class\"\" LIMIT 0; UPDATE t SET v='x' WHERE id=1; SELECT * FROM \"\"pg_class\"", + AbstractDBSpecificConnector.quoteIdentifier(payload, '"')); + } +} diff --git a/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java b/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java new file mode 100644 index 000000000..6e04ede7c --- /dev/null +++ b/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java @@ -0,0 +1,136 @@ +/* + * 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 org.junit.Assert; +import org.junit.Test; + +/** + * Unit tests for {@link SelectQueryValidator} + */ +public class SelectQueryValidatorTest { + + @Test + public void testValidSelectQueries() { + String[] queries = { + "SELECT * FROM my_table", + "select id, name from my_table where id > 10 order by name", + "SELECT * FROM my_table WHERE $CONDITIONS", + "SELECT * FROM my_table WHERE id > 10 AND $CONDITIONS", + "SELECT MIN(id), MAX(id) FROM my_table", + "SELECT * FROM \"my schema\".\"my;table\"", + "SELECT * FROM \"pg_class\"\"; DELETE FROM t; --\"", + "SELECT * FROM `db`.`my;table`", + "SELECT a.id, b.name FROM a JOIN b ON a.id = b.id", + "SELECT * FROM (SELECT id FROM my_table) t", + "SELECT id FROM a UNION ALL SELECT id FROM b", + "WITH t AS (SELECT id FROM my_table) SELECT * FROM t", + "SELECT * FROM my_table WHERE name = 'a;b' -- trailing ; comment", + "SELECT * FROM my_table /* ; DELETE FROM t */", + "SELECT * FROM my_table;", + "SELECT TOP 10 * FROM dbo.my_table", + "SELECT * FROM my_table WHERE ROWNUM <= 10", + "SELECT * FROM my_table LIMIT 10", + }; + for (String query : queries) { + try { + SelectQueryValidator.validate(query, false); + } catch (IllegalArgumentException e) { + Assert.fail(String.format("Query '%s' should be valid but failed with: %s", query, e.getMessage())); + } + } + } + + @Test + public void testSquareBracketIdentifiers() { + SelectQueryValidator.validate("SELECT * FROM [dbo].[my;table] WHERE [id] > 1", true); + try { + SelectQueryValidator.validate("SELECT * FROM [dbo].[t]; DELETE FROM [dbo].[t]", true); + Assert.fail("Stacked statements with square bracket identifiers should have been rejected"); + } catch (IllegalArgumentException e) { + // expected + } + } + + @Test + public void testInvalidQueries() { + String[] queries = { + // stacked statements, as in the reported vulnerability + "SELECT * FROM \"pg_catalog\".\"pg_class\" LIMIT 0; UPDATE t SET v = 'x' WHERE id = 1; SELECT 1", + "SELECT * FROM my_table; DELETE FROM my_table", + "SELECT 1; SELECT 2", + // data modification and definition + "UPDATE my_table SET v = 1", + "DELETE FROM my_table", + "INSERT INTO my_table VALUES (1)", + "DROP TABLE my_table", + "TRUNCATE TABLE my_table", + "CREATE TABLE t (id INT)", + "ALTER TABLE my_table ADD COLUMN c INT", + "MERGE INTO t USING s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.v = s.v", + "GRANT SELECT ON my_table TO someone", + // session and database commands + "SET search_path TO other", + "USE other_db", + "ALTER SESSION SET NLS_DATE_FORMAT = 'YYYY-MM-DD'", + // stored procedures + "EXEC my_procedure", + "CALL my_procedure()", + // reads that write or lock + "SELECT * INTO new_table FROM my_table", + "SELECT * FROM my_table FOR UPDATE", + "WITH d AS (DELETE FROM my_table RETURNING *) SELECT * FROM d", + "SELECT * FROM my_table INTO OUTFILE '/tmp/out.csv'", + "SELECT * INTO OUTFILE '/tmp/out.csv' FROM my_table", + // a dollar quoted string (PostgreSQL) that hides a second statement + "SELECT $$ ' $$; DELETE FROM my_table; -- '", + // a string that hides a second statement when backslash escapes quotes (MySQL) but not otherwise + "SELECT 'a\\''; DELETE FROM my_table; -- '", + // a string that hides a second statement when backslash does not escape quotes (PostgreSQL) + "SELECT 'a\\'; DELETE FROM my_table; --'", + // empty and unparsable + "", + ";", + "SELECT FROM WHERE", + }; + for (String query : queries) { + try { + SelectQueryValidator.validate(query, false); + Assert.fail(String.format("Query '%s' should have been rejected", query)); + } catch (IllegalArgumentException e) { + // expected + } + } + } + + @Test + public void testErrorMessages() { + assertError("SELECT 1; DELETE FROM t", "Only a single SELECT statement is allowed, but found 2 statements."); + assertError("DELETE FROM t", "Only a SELECT statement is allowed, but found a Delete statement."); + assertError("SELECT * INTO t2 FROM t", "SELECT ... INTO is not allowed, the query must only read data."); + assertError("SELECT * FROM t FOR UPDATE", "SELECT ... FOR UPDATE is not allowed, the query must only read data."); + } + + private static void assertError(String query, String expectedMessage) { + try { + SelectQueryValidator.validate(query, false); + Assert.fail(String.format("Query '%s' should have been rejected", query)); + } catch (IllegalArgumentException e) { + Assert.assertEquals(expectedMessage, e.getMessage()); + } + } +} diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java index c78ceca6f..5997cf627 100644 --- a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java +++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java @@ -116,15 +116,15 @@ protected SchemaReader getSchemaReader(String sessionID) { @Override protected String getTableName(String database, String schema, String table) { if (database == null && schema == null) { - return String.format("`%s`", table); + return quoteIdentifier(table, '`'); } if (database == null) { - return String.format("`%s`.`%s`", schema, table); + return quoteIdentifier(schema, '`') + "." + quoteIdentifier(table, '`'); } if (schema == null) { - return String.format("`%s`.`%s`", database, table); + return quoteIdentifier(database, '`') + "." + quoteIdentifier(table, '`'); } - return String.format("`%s`.`%s`.`%s`", database, schema, table); + return quoteIdentifier(database, '`') + "." + quoteIdentifier(schema, '`') + "." + quoteIdentifier(table, '`'); } @Override @@ -136,6 +136,7 @@ protected String getRandomQuery(String tableName, int limit) { @Override protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) { + String strataColumn = quoteIdentifier(strata, '`'); return String.format("WITH t_%s AS (\n" + " SELECT *,\n" + " ROW_NUMBER() OVER (ORDER BY %s, RAND()) AS sqn_%s,\n" + @@ -146,8 +147,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata, "WHERE MOD(sqn_%s, GREATEST(1, CAST(c_%s / %d AS BIGINT))) = 1\n" + "ORDER BY %s\n" + "LIMIT %d", - sessionID, strata, sessionID, sessionID, tableName, sessionID, sessionID, sessionID, - limit, strata, limit); + sessionID, strataColumn, sessionID, sessionID, tableName, sessionID, sessionID, sessionID, + limit, strataColumn, limit); } @Override diff --git a/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java index a04dfb398..27b830c76 100644 --- a/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java +++ b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java @@ -38,6 +38,21 @@ public void testGetTableName() { CONNECTOR.getTableName(null, null, "my_table")); } + @Test + public void testGetTableNameEscapesBackticks() { + Assert.assertEquals("`main`.`default`.`t``; DELETE FROM x; --`", + CONNECTOR.getTableName("main", "default", "t`; DELETE FROM x; --")); + Assert.assertEquals("`my``schema`.`my``table`", + CONNECTOR.getTableName(null, "my`schema", "my`table")); + } + + @Test + public void testGetStratifiedQueryEscapesStrata() { + String query = CONNECTOR.getStratifiedQuery("`main`.`default`.`my_table`", 10, "id`; DROP TABLE t; --", "s1"); + Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`, RAND()")); + Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`\n")); + } + @Test public void testGetRandomQuery() { Assert.assertEquals("SELECT * FROM `main`.`default`.`my_table`\n" + @@ -49,13 +64,13 @@ public void testGetRandomQuery() { public void testGetStratifiedQuery() { Assert.assertEquals("WITH t_s1 AS (\n" + " SELECT *,\n" + - " ROW_NUMBER() OVER (ORDER BY id, RAND()) AS sqn_s1,\n" + + " ROW_NUMBER() OVER (ORDER BY `id`, RAND()) AS sqn_s1,\n" + " COUNT(*) OVER () AS c_s1\n" + " FROM `main`.`default`.`my_table`\n" + " )\n" + "SELECT * FROM t_s1\n" + "WHERE MOD(sqn_s1, GREATEST(1, CAST(c_s1 / 10 AS BIGINT))) = 1\n" + - "ORDER BY id\n" + + "ORDER BY `id`\n" + "LIMIT 10", CONNECTOR.getStratifiedQuery("`main`.`default`.`my_table`", 10, "id", "s1")); } diff --git a/generic-database-plugin/src/main/java/io/cdap/plugin/jdbc/DatabaseSource.java b/generic-database-plugin/src/main/java/io/cdap/plugin/jdbc/DatabaseSource.java index b35b025b7..8240ffee3 100644 --- a/generic-database-plugin/src/main/java/io/cdap/plugin/jdbc/DatabaseSource.java +++ b/generic-database-plugin/src/main/java/io/cdap/plugin/jdbc/DatabaseSource.java @@ -23,6 +23,7 @@ import io.cdap.plugin.db.ConnectionConfig; import io.cdap.plugin.db.source.AbstractDBSource; +import java.util.Locale; import javax.annotation.Nullable; /** @@ -46,6 +47,18 @@ protected String createConnectionString() { return databaseSourceConfig.connectionString; } + @Override + protected boolean useSquareBracketQuotes() { + // SQL Server quotes identifiers with square brackets + String connectionString = databaseSourceConfig.connectionString; + if (connectionString == null) { + return false; + } + String lowerCaseConnectionString = connectionString.toLowerCase(Locale.ROOT); + return lowerCaseConnectionString.startsWith("jdbc:sqlserver:") + || lowerCaseConnectionString.startsWith("jdbc:jtds:sqlserver:"); + } + /** * Generic database source configuration. */ diff --git a/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java b/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java index bedb720a2..f11342461 100644 --- a/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java +++ b/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java @@ -20,16 +20,19 @@ import io.cdap.cdap.api.common.Bytes; import io.cdap.cdap.api.data.format.StructuredRecord; import io.cdap.cdap.api.dataset.table.Table; +import io.cdap.cdap.datapipeline.SmartWorkflow; import io.cdap.cdap.etl.api.batch.BatchSource; import io.cdap.cdap.etl.mock.batch.MockSink; import io.cdap.cdap.etl.proto.v2.ETLBatchConfig; import io.cdap.cdap.etl.proto.v2.ETLPlugin; import io.cdap.cdap.etl.proto.v2.ETLStage; +import io.cdap.cdap.proto.ProgramRunStatus; import io.cdap.cdap.proto.artifact.AppRequest; import io.cdap.cdap.proto.id.ApplicationId; import io.cdap.cdap.proto.id.NamespaceId; import io.cdap.cdap.test.ApplicationManager; import io.cdap.cdap.test.DataSetManager; +import io.cdap.cdap.test.WorkflowManager; import io.cdap.plugin.GenericDatabasePluginTestBase; import io.cdap.plugin.common.Constants; import io.cdap.plugin.db.ConnectionConfig; @@ -41,7 +44,11 @@ import java.math.BigDecimal; import java.math.MathContext; import java.nio.ByteBuffer; +import java.sql.Connection; import java.sql.Date; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; import java.sql.Time; import java.text.SimpleDateFormat; import java.time.LocalDate; @@ -52,6 +59,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; /** * Test for ETL using databases. @@ -329,6 +337,61 @@ public void testUserNamePasswordCombinations() throws Exception { deployApplication(appId, appRequest); } + @Test + public void testNonSelectImportQueryFailsDeployment() throws Exception { + int rowCount = countRows("my_table"); + String importQuery = "SELECT * FROM \"my_table\" WHERE $CONDITIONS; DELETE FROM \"my_table\""; + ETLPlugin sourceConfig = new ETLPlugin( + DatabaseConstants.PLUGIN_NAME, + BatchSource.PLUGIN_TYPE, + ImmutableMap.builder() + .put(ConnectionConfig.CONNECTION_STRING, getConnectionURL()) + .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME) + .put(AbstractDBSource.DBSourceConfig.IMPORT_QUERY, importQuery) + .put(AbstractDBSource.DBSourceConfig.BOUNDING_QUERY, "SELECT MIN(ID),MAX(ID) from \"my_table\"") + .put(AbstractDBSource.DBSourceConfig.SPLIT_BY, "ID") + .put(Constants.Reference.REFERENCE_NAME, "DBNonSelectQueryTest") + .build(), + null); + ETLBatchConfig etlConfig = getETLBatchConfig(sourceConfig, MockSink.getPlugin("nonSelectQueryOutput")); + assertDeploymentFailure(NamespaceId.DEFAULT.app("dbSourceNonSelectQueryTest"), etlConfig, DATAPIPELINE_ARTIFACT, + "Deploying DB Source with a non SELECT import query should have failed."); + Assert.assertEquals(rowCount, countRows("my_table")); + } + + @Test + public void testNonSelectImportQueryMacroFailsRun() throws Exception { + int rowCount = countRows("my_table"); + ETLPlugin sourceConfig = new ETLPlugin( + DatabaseConstants.PLUGIN_NAME, + BatchSource.PLUGIN_TYPE, + ImmutableMap.builder() + .put(ConnectionConfig.CONNECTION_STRING, getConnectionURL()) + .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME) + .put(AbstractDBSource.DBSourceConfig.IMPORT_QUERY, "${importQuery}") + .put(AbstractDBSource.DBSourceConfig.BOUNDING_QUERY, "SELECT MIN(ID),MAX(ID) from \"my_table\"") + .put(AbstractDBSource.DBSourceConfig.SPLIT_BY, "ID") + .put(Constants.Reference.REFERENCE_NAME, "DBNonSelectQueryMacroTest") + .build(), + null); + ApplicationManager appManager = deployETL(sourceConfig, MockSink.getPlugin("nonSelectQueryMacroOutput"), + DATAPIPELINE_ARTIFACT, "dbSourceNonSelectQueryMacroTest"); + WorkflowManager workflowManager = appManager.getWorkflowManager(SmartWorkflow.NAME); + workflowManager.start(ImmutableMap.of( + "importQuery", "SELECT * FROM \"my_table\" WHERE $CONDITIONS; DELETE FROM \"my_table\"")); + workflowManager.waitForRun(ProgramRunStatus.FAILED, 5, TimeUnit.MINUTES); + Assert.assertEquals(rowCount, countRows("my_table")); + } + + private int countRows(String table) throws SQLException { + try (Connection connection = getConnection(); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(String.format("SELECT COUNT(*) FROM \"%s\"", table))) { + resultSet.next(); + return resultSet.getInt(1); + } + } + @Test public void testNonExistentDBTable() throws Exception { // source diff --git a/generic-database-plugin/src/test/java/io/cdap/plugin/jdbc/DatabaseSourceTest.java b/generic-database-plugin/src/test/java/io/cdap/plugin/jdbc/DatabaseSourceTest.java new file mode 100644 index 000000000..c03a64a85 --- /dev/null +++ b/generic-database-plugin/src/test/java/io/cdap/plugin/jdbc/DatabaseSourceTest.java @@ -0,0 +1,42 @@ +/* + * 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.jdbc; + +import org.junit.Assert; +import org.junit.Test; + +/** + * Unit tests for {@link DatabaseSource} + */ +public class DatabaseSourceTest { + + @Test + public void testUseSquareBracketQuotes() { + Assert.assertTrue(createSource("jdbc:sqlserver://localhost:1433;databaseName=db").useSquareBracketQuotes()); + Assert.assertTrue(createSource("jdbc:jtds:sqlserver://localhost:1433/db").useSquareBracketQuotes()); + Assert.assertTrue(createSource("JDBC:SQLSERVER://localhost:1433").useSquareBracketQuotes()); + Assert.assertFalse(createSource("jdbc:postgresql://localhost:5432/db").useSquareBracketQuotes()); + Assert.assertFalse(createSource("jdbc:mysql://localhost:3306/db").useSquareBracketQuotes()); + Assert.assertFalse(createSource(null).useSquareBracketQuotes()); + } + + private static DatabaseSource createSource(String connectionString) { + DatabaseSource.DatabaseSourceConfig config = new DatabaseSource.DatabaseSourceConfig(); + config.connectionString = connectionString; + return new DatabaseSource(config); + } +} diff --git a/mssql-plugin/src/e2e-test/resources/errorMessage.properties b/mssql-plugin/src/e2e-test/resources/errorMessage.properties index 1f328f4d6..f94faf41e 100644 --- a/mssql-plugin/src/e2e-test/resources/errorMessage.properties +++ b/mssql-plugin/src/e2e-test/resources/errorMessage.properties @@ -22,6 +22,6 @@ errorMessageInvalidsourcetable=Spark program 'phase-1' failed with error: Stage errorMessage: SQL Exception occurred: [Message='Incorrect syntax near the keyword 'table'.', SQLState='S0001', ErrorCode='156'].. Please check the system logs for more details. errorMessageInvalidCredentialSource=Spark program 'phase-1' failed with error: Stage 'SQL Server' encountered : java.lang.IllegalArgumentException: \ Plugin with id SQL Server:source.jdbc.sqlserver does not exist in program phase-1 of application -errorLogsMessageInvalidBoundingQuery=Spark program 'phase-1' failed with error: Stage 'SQL Server' encountered : java.io.IOException: Could not find stored procedure +errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid Bounding Query blank.jdbcPluginName.message=Required property 'jdbcPluginName' has no value. blank.connection.message=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: 'S0001', errorCode: '18456', errorMessage: SQL Exception occurred: [Message='Login failed for user ' diff --git a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerConnector.java b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerConnector.java index 7e86d97ce..4f3ba8963 100644 --- a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerConnector.java +++ b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerConnector.java @@ -117,7 +117,8 @@ public StructuredRecord transform(LongWritable longWritable, SqlServerSourceDBRe @Override protected String getTableName(String database, String schema, String table) { - return String.format("\"%s\".\"%s\".\"%s\"", database, schema, table); + return quoteIdentifier(database, '"') + "." + quoteIdentifier(schema, '"') + "." + + quoteIdentifier(table, '"'); } @Override @@ -137,6 +138,7 @@ protected String getRandomQuery(String tableName, int limit) { @Override protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) { + String strataColumn = quoteIdentifier(strata, '"'); return String.format("WITH t_%s AS (\n" + " SELECT *,\n" + " ROW_NUMBER() OVER (ORDER BY %s, RAND()) AS sqn_%s,\n" + @@ -146,8 +148,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata, "SELECT TOP %d * FROM t_%s\n" + "WHERE sqn_%s %% CAST(0.5 * ((c_%s / %d + 1) + ABS(c_%s / %d - 1)) AS bigint) = 1\n" + "ORDER BY %s", - sessionID, strata, sessionID, sessionID, tableName, limit, sessionID, sessionID, sessionID, - limit, sessionID, limit, strata); + sessionID, strataColumn, sessionID, sessionID, tableName, limit, sessionID, sessionID, + sessionID, limit, sessionID, limit, strataColumn); } @Override diff --git a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java index d8d4cf39e..b6bcb9ff9 100644 --- a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java +++ b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java @@ -81,6 +81,11 @@ protected String getErrorDetailsProviderClassName() { return SqlServerErrorDetailsProvider.class.getName(); } + @Override + protected boolean useSquareBracketQuotes() { + return true; + } + @Override protected DBErrorDetailsProvider getErrorDetailsProvider() { if (dbErrorDetailsProvider == null) { diff --git a/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java b/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java index 0163cd55e..231d072b9 100644 --- a/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java +++ b/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java @@ -34,6 +34,27 @@ public class SqlServerConnectorUnitTest { // private static final SqlServerConnector CONNECTOR = new SqlServerConnector(new SqlServerConnectorConfig()); + /** + * Unit test for getTableName() with identifiers containing double quotes + */ + @Test + public void getTableNameEscapesQuotesTest() { + Assert.assertEquals("\"db\".\"dbo\".\"t\"\"; DELETE FROM x; --\"", + CONNECTOR.getTableName("db", "dbo", "t\"; DELETE FROM x; --")); + Assert.assertEquals("\"my\"\"db\".\"my\"\"schema\".\"my\"\"table\"", + CONNECTOR.getTableName("my\"db", "my\"schema", "my\"table")); + } + + /** + * Unit test for getStratifiedQuery() with a strata column containing double quotes + */ + @Test + public void getStratifiedQueryEscapesStrataTest() { + String query = CONNECTOR.getStratifiedQuery("\"db\".\"dbo\".\"table\"", 10, "id\"; DROP TABLE t; --", "s1"); + Assert.assertTrue(query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\", RAND()")); + Assert.assertTrue(query.endsWith("ORDER BY \"id\"\"; DROP TABLE t; --\"")); + } + /** * Unit tests for getTableQuery() */ diff --git a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConnector.java b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConnector.java index 214d8650e..0f854bc88 100644 --- a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConnector.java +++ b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConnector.java @@ -98,7 +98,7 @@ protected void setConnectorSpec(ConnectorSpecRequest request, DBConnectorPath pa @Override protected String getTableName(String database, String schema, String table) { - return String.format("`%s`.`%s`", database, table); + return quoteIdentifier(database, '`') + "." + quoteIdentifier(table, '`'); } @Override @@ -112,6 +112,7 @@ protected String getRandomQuery(String tableName, int limit) { @Override protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) { + String strataColumn = quoteIdentifier(strata, '`'); return String.format("WITH t_%s AS (\n" + " SELECT *,\n" + " ROW_NUMBER() OVER (ORDER BY %s, RAND()) AS sqn_%s,\n" + @@ -122,8 +123,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata, "WHERE MOD(sqn_%s, GREATEST(1, CAST(c_%s / %d AS UNSIGNED))) = 1\n" + "ORDER BY %s\n" + "LIMIT %d", - sessionID, strata, sessionID, sessionID, tableName, sessionID, sessionID, sessionID, - limit, strata, limit); + sessionID, strataColumn, sessionID, sessionID, tableName, sessionID, sessionID, sessionID, + limit, strataColumn, limit); } @Override diff --git a/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java b/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java index df731d194..01820842e 100644 --- a/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java +++ b/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java @@ -42,6 +42,25 @@ public void getTableNameTest() { CONNECTOR.getTableName("db", "schema", "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", "schema", "t`; DELETE FROM x; --")); + } + + /** + * Unit test for getStratifiedQuery() with a strata column containing backticks + */ + @Test + public void getStratifiedQueryEscapesStrataTest() { + String query = CONNECTOR.getStratifiedQuery("`db`.`table`", 10, "id`; DROP TABLE t; --", "s1"); + Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`, RAND()")); + Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`\n")); + } + /** * Unit tests for getTableQuery() */ diff --git a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java index 469ca45f2..a59c0688e 100644 --- a/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java +++ b/oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleConnector.java @@ -135,7 +135,7 @@ protected String getConnectionString(@Nullable String database) { @Override protected String getTableName(String database, String schema, String table) { - return String.format("\"%s\".\"%s\"", schema, table); + return quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"'); } @Override @@ -157,6 +157,7 @@ protected String getRandomQuery(String tableName, int limit) { @Override protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) { + String strataColumn = quoteIdentifier(strata, '"'); return String.format("WITH t_%s AS (\n" + " SELECT %s.*,\n" + " ROW_NUMBER() OVER (ORDER BY %s, dbms_random.random) AS s_%s,\n" + @@ -166,8 +167,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata, " )\n" + "SELECT * FROM t_%s\n" + "WHERE MOD(s_%s, GREATEST(1, CAST(c_%s / %d AS INTEGER))) = 1 AND ROWNUM <= %d", - sessionID, tableName, strata, sessionID, tableName, sessionID, tableName, - strata, sessionID, sessionID, sessionID, limit, limit); + sessionID, tableName, strataColumn, sessionID, tableName, sessionID, tableName, + strataColumn, sessionID, sessionID, sessionID, limit, limit); } @Override diff --git a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java index f89a6c1d7..2441862d0 100644 --- a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java +++ b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java @@ -37,6 +37,25 @@ 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\"\" WHERE 1=1 --\"", + CONNECTOR.getTableName("db", "MY\"SCHEMA", "T\" WHERE 1=1 --")); + } + + /** + * Unit test for getStratifiedQuery() with a strata column containing double quotes + */ + @Test + public void getStratifiedQueryEscapesStrataTest() { + String query = CONNECTOR.getStratifiedQuery("\"SCHEMA\".\"TABLE\"", 10, "ID\", 1) --", "s1"); + Assert.assertTrue(query.contains("ORDER BY \"ID\"\", 1) --\", dbms_random.random")); + Assert.assertTrue(query.contains("ORDER BY \"ID\"\", 1) --\" )")); + } + /** * Unit tests for getTableQuery() */ diff --git a/pom.xml b/pom.xml index c739d8855..a04552767 100644 --- a/pom.xml +++ b/pom.xml @@ -310,6 +310,12 @@ failsafe 3.3.2 + + + com.github.jsqlparser + jsqlparser + 4.9 + org.hsqldb hsqldb diff --git a/postgresql-plugin/src/e2e-test/resources/errorMessage.properties b/postgresql-plugin/src/e2e-test/resources/errorMessage.properties index a6c500f1d..9dacc0b67 100644 --- a/postgresql-plugin/src/e2e-test/resources/errorMessage.properties +++ b/postgresql-plugin/src/e2e-test/resources/errorMessage.properties @@ -18,5 +18,4 @@ errorMessageInvalidSourceHost=SQL error while getting query schema: The connecti 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 errorMessageInvalidHost=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '08001', errorCode: '0', errorMessage: SQL Exception occurred: [Message='The connection attempt failed.', SQLState='08001', ErrorCode='0'].' -errorLogsMessageInvalidBoundingQuery=Spark program 'phase-1' failed with error: Stage 'PostgreSQL' encountered : \ - java.io.IOException: The column index is out of range: 1, number of columns: 0.. Please check the system logs for more details. +errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid diff --git a/postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresConnector.java b/postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresConnector.java index 5b23ce6cd..4ad703b61 100644 --- a/postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresConnector.java +++ b/postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresConnector.java @@ -117,7 +117,7 @@ public StructuredRecord transform(LongWritable longWritable, PostgresDBRecord re @Override protected String getTableName(String database, String schema, String table) { - return String.format("\"%s\".\"%s\"", schema, table); + return quoteIdentifier(schema, '"') + "." + quoteIdentifier(table, '"'); } @Override @@ -129,6 +129,7 @@ protected String getRandomQuery(String tableName, int limit) { @Override protected String getStratifiedQuery(String tableName, int limit, String strata, String sessionID) { + String strataColumn = quoteIdentifier(strata, '"'); // this query times out on larger datasets; would benefit from optimization return String.format("WITH t_%s AS (\n" + " SELECT *,\n" + @@ -140,8 +141,8 @@ protected String getStratifiedQuery(String tableName, int limit, String strata, "WHERE mod(sqn_%s, GREATEST(1, CAST(c_%s / %d AS bigint))) = 1\n" + "ORDER BY %s\n" + "LIMIT %d", - sessionID, strata, sessionID, sessionID, tableName, sessionID, sessionID, sessionID, - limit, strata, limit); + sessionID, strataColumn, sessionID, sessionID, tableName, sessionID, sessionID, sessionID, + limit, strataColumn, limit); } @Override diff --git a/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java b/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java index f286cf574..57e6399d4 100644 --- a/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java +++ b/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java @@ -41,6 +41,28 @@ public void getTableNameTest() { CONNECTOR.getTableName("db", "schema", "table")); } + /** + * Unit test for getTableName() with identifiers containing double quotes + */ + @Test + public void getTableNameEscapesQuotesTest() { + Assert.assertEquals("\"pg_catalog\".\"pg_class\"\" LIMIT 0; UPDATE t SET v='x'; SELECT * FROM \"\"pg_class\"", + CONNECTOR.getTableName("db", "pg_catalog", + "pg_class\" LIMIT 0; UPDATE t SET v='x'; SELECT * FROM \"pg_class")); + Assert.assertEquals("\"my\"\"schema\".\"my\"\"table\"", + CONNECTOR.getTableName("db", "my\"schema", "my\"table")); + } + + /** + * Unit test for getStratifiedQuery() with a strata column containing double quotes + */ + @Test + public void getStratifiedQueryEscapesStrataTest() { + String query = CONNECTOR.getStratifiedQuery("\"schema\".\"table\"", 10, "id\"; DROP TABLE t; --", "s1"); + Assert.assertTrue(query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\", random()")); + Assert.assertTrue(query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\"\n")); + } + /** * Unit tests for getTableQuery() */ From 7614b6c4ee2a3e8434c8c99907fd519f0acaa46c Mon Sep 17 00:00:00 2001 From: vishwasvaidya-cloudsufi Date: Tue, 10 Nov 2026 09:50:20 +0530 Subject: [PATCH 2/2] harden source-query validation and fix false positives --- .../auroradb/mysql/AuroraMysqlSource.java | 6 + .../resources/errorMessage.properties | 2 +- .../resources/pluginParameters.properties | 2 +- .../cloudsql/mysql/CloudSQLMySQLSource.java | 6 + .../resources/errorMessage.properties | 2 +- .../resources/pluginParameters.properties | 2 +- .../AbstractDBSpecificConnector.java | 4 +- .../plugin/db/source/AbstractDBSource.java | 63 ++- .../plugin/util/SelectQueryValidator.java | 371 +++++++++++++++--- .../plugin/util/SelectQueryValidatorTest.java | 202 +++++++--- .../databricks/DatabricksConnector.java | 17 +- .../plugin/databricks/DatabricksSource.java | 6 + .../DatabricksConnectorUnitTest.java | 4 +- .../cdap/plugin/db/sink/DBSourceTestRun.java | 50 +++ .../io/cdap/plugin/mariadb/MariadbSource.java | 6 + .../plugin/memsql/source/MemsqlSource.java | 6 + .../resources/pluginParameters.properties | 1 + .../io/cdap/plugin/mssql/SqlServerAction.java | 3 +- .../plugin/mssql/SqlServerPostAction.java | 3 +- .../io/cdap/plugin/mssql/SqlServerSink.java | 3 +- .../io/cdap/plugin/mssql/SqlServerSource.java | 3 +- .../mssql/SqlServerConnectorUnitTest.java | 4 +- .../io/cdap/plugin/mysql/MysqlAction.java | 3 +- .../io/cdap/plugin/mysql/MysqlPostAction.java | 3 +- .../java/io/cdap/plugin/mysql/MysqlSink.java | 3 +- .../io/cdap/plugin/mysql/MysqlSource.java | 9 +- .../plugin/mysql/MysqlConnectorUnitTest.java | 4 +- .../resources/errorMessage.properties | 3 +- .../resources/pluginParameters.properties | 2 +- .../oracle/OracleConnectorUnitTest.java | 4 +- .../resources/errorMessage.properties | 2 +- .../resources/pluginParameters.properties | 2 +- .../postgres/PostgresConnectorUnitTest.java | 4 +- 33 files changed, 656 insertions(+), 149 deletions(-) diff --git a/aurora-mysql-plugin/src/main/java/io/cdap/plugin/auroradb/mysql/AuroraMysqlSource.java b/aurora-mysql-plugin/src/main/java/io/cdap/plugin/auroradb/mysql/AuroraMysqlSource.java index 955d0be9e..67d962cb9 100644 --- a/aurora-mysql-plugin/src/main/java/io/cdap/plugin/auroradb/mysql/AuroraMysqlSource.java +++ b/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. */ diff --git a/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties b/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties index f6ed2205e..92ed7ddbf 100644 --- a/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties +++ b/cloudsql-mysql-plugin/src/e2e-test/resources/errorMessage.properties @@ -18,7 +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=Errors were encountered during validation. Invalid +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 ' diff --git a/cloudsql-mysql-plugin/src/e2e-test/resources/pluginParameters.properties b/cloudsql-mysql-plugin/src/e2e-test/resources/pluginParameters.properties index 04fdc9ece..775110a42 100644 --- a/cloudsql-mysql-plugin/src/e2e-test/resources/pluginParameters.properties +++ b/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 diff --git a/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLSource.java b/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLSource.java index 43c63aa9e..612b55215 100644 --- a/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLSource.java +++ b/cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLSource.java @@ -83,6 +83,12 @@ protected Class 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( diff --git a/cloudsql-postgresql-plugin/src/e2e-test/resources/errorMessage.properties b/cloudsql-postgresql-plugin/src/e2e-test/resources/errorMessage.properties index a344e472f..cb99c64ed 100644 --- a/cloudsql-postgresql-plugin/src/e2e-test/resources/errorMessage.properties +++ b/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 :: to connect to \ a public CloudSQL PostgreSQL instance. errorMessagePrivateConnectionName=Enter the internal IP address of the Compute Engine VM cloudsql proxy is running on, \ diff --git a/cloudsql-postgresql-plugin/src/e2e-test/resources/pluginParameters.properties b/cloudsql-postgresql-plugin/src/e2e-test/resources/pluginParameters.properties index f619e5f3c..072bb29ec 100644 --- a/cloudsql-postgresql-plugin/src/e2e-test/resources/pluginParameters.properties +++ b/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 diff --git a/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java b/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java index f1b0fdf6f..5637aa326 100644 --- a/database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java +++ b/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; @@ -158,7 +159,8 @@ protected String getTableName(String database, String schema, String 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) { + 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; } diff --git a/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java b/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java index 9517bb2b9..f3e7711b1 100644 --- a/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java +++ b/database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java @@ -107,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; @@ -120,7 +124,8 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) { Class 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) { @@ -162,6 +167,9 @@ public Schema getSchema(Class 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(); @@ -169,14 +177,19 @@ 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 + private void validateImportQuery(String query) { + String finalQuery = query != null && query.contains("$CONDITIONS") ? removeConditionsClause(query) : query; try { - SelectQueryValidator.validate(finalQuery, useSquareBracketQuotes()); + 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) retryPolicy, connection, getErrorDetailsProvider())) { statement.setMaxRows(1); @@ -250,14 +263,16 @@ protected boolean useSquareBracketQuotes() { 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) { - if (Strings.isNullOrEmpty(query)) { + // 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()); + 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.") @@ -265,6 +280,40 @@ private void validateQuery(String property, String label, @Nullable String query } } + /** + * 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. diff --git a/database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java b/database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java index 9305d84a1..ae399e627 100644 --- a/database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java +++ b/database-commons/src/main/java/io/cdap/plugin/util/SelectQueryValidator.java @@ -18,110 +18,381 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder; import net.sf.jsqlparser.JSQLParserException; +import net.sf.jsqlparser.expression.Expression; import net.sf.jsqlparser.parser.CCJSqlParserUtil; +import net.sf.jsqlparser.schema.Column; 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.Arrays; +import java.util.HashSet; import java.util.List; import java.util.Objects; +import java.util.Set; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; import java.util.stream.Collectors; +import javax.annotation.Nullable; /** * Validates that a user provided query is a single, read-only SELECT statement before it is executed with the * credential of a database connection. + * + *

The check is primarily lexical: the query is scanned with awareness of string literals, comments and quoted + * identifiers, and is required to be a single statement that reads (contains {@code SELECT}) and carries no write or + * data-definition keyword, no {@code INTO}, and no {@code FOR UPDATE}/{@code SHARE} locking clause outside of quoted + * text. A lexical scan is used because no single SQL parser understands every dialect the connectors support, and a + * parser that mis-lexes a construct (dollar quotes, executable comments, dialect-specific clauses) either lets an + * injection through or rejects a valid query. A few constructs that parsers are known to lex differently from the + * database driver are rejected outright up front. JSqlParser is still run as a best-effort secondary check, but only + * to add confidence when it succeeds; a query it cannot parse is not rejected on that basis alone. */ 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(); + // A single shared, bounded pool of daemon threads for the parser, instead of two new executors per call. The + // unbounded queue makes concurrent bursts wait for a thread rather than being rejected, and idle threads still + // terminate after the keep-alive. + private static final ExecutorService PARSER_EXECUTOR = createParserExecutor(); + + private static ExecutorService createParserExecutor() { + int maxThreads = Math.max(2, Runtime.getRuntime().availableProcessors() * 2); + ThreadPoolExecutor executor = new ThreadPoolExecutor( + maxThreads, maxThreads, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), PARSER_THREAD_FACTORY); + executor.allowCoreThreadTimeOut(true); + return executor; + } + + // Constructs that JSqlParser and the database drivers lex differently, so a crafted literal can look like one + // SELECT to the parser while the driver splits it into several statements. They never occur in a sampling query. + private static final Pattern DOLLAR_QUOTE = + Pattern.compile("\\$([A-Za-z_\\u0080-\\uFFFF][A-Za-z0-9_\\u0080-\\uFFFF]*)?\\$"); + private static final Pattern ORACLE_QUOTE = Pattern.compile("(^|[^A-Za-z0-9_])[qQ]'", Pattern.MULTILINE); + private static final Pattern EXECUTABLE_COMMENT = Pattern.compile("/\\*(M)?!"); + + // Keywords that, used as a statement (not as a function call), write or change schema. Matched only when the + // keyword is not immediately followed by '(' so legitimate functions such as MySQL TRUNCATE(x, d) are allowed. + private static final Set WRITE_KEYWORDS = new HashSet<>(Arrays.asList( + "INSERT", "UPDATE", "DELETE", "MERGE", "UPSERT", "CREATE", "DROP", "ALTER", "TRUNCATE", "RENAME", + "GRANT", "REVOKE", "CALL", "EXEC", "EXECUTE", "INTO")); + // Of the write keywords, only these also name scalar functions (e.g. MySQL TRUNCATE(x, d), INSERT(s, p, l, n)). + // A trailing '(' exempts a write keyword only for these, and only when it is not the first word of the statement, + // so statements such as UPDATE (SELECT ...) SET ... or EXEC (...) are never allowed through. + private static final Set FUNCTION_COLLIDING_WRITE_KEYWORDS = + new HashSet<>(Arrays.asList("TRUNCATE", "INSERT")); + // Tokens that, following FOR, acquire locks (FOR UPDATE / FOR SHARE / FOR NO KEY UPDATE). FOR READ ONLY and + // FOR FETCH ONLY are read-only and allowed. + private static final Set LOCKING_AFTER_FOR = new HashSet<>(Arrays.asList("UPDATE", "SHARE", "NO")); private SelectQueryValidator() { } /** - * Validates that the given query is a single read-only SELECT statement. + * Validates that the given query is a single read-only SELECT statement, trying both backslash-escape + * interpretations since the dialect is unknown. * * @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 + * @throws IllegalArgumentException if the query is not a single read-only SELECT statement */ public static void validate(String query, boolean squareBracketQuotes) { + validate(query, squareBracketQuotes, null); + } + + /** + * 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 + * @param backslashEscapes whether a backslash escapes a quote inside a string literal in the database's dialect; + * {@code null} when the dialect is unknown, in which case the query must pass under both interpretations + * @throws IllegalArgumentException if the query is not a single read-only SELECT statement + */ + public static void validate(String query, boolean squareBracketQuotes, @Nullable Boolean backslashEscapes) { + if (query == null) { + throw new IllegalArgumentException("Query must not be null."); + } 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); + rejectLexerMismatchConstructs(parsableQuery); + // A double quote is a string literal in MySQL/MariaDB/SingleStore/Databricks by default (where a backslash can + // escape it) but an identifier when ANSI_QUOTES is enabled (where it cannot). When the dialect is unknown, or it + // is a backslash dialect and the query contains a double quote, require the query to be a single read-only SELECT + // under both backslash interpretations so a crafted literal cannot break out either way. + boolean bothModes = backslashEscapes == null + || (Boolean.TRUE.equals(backslashEscapes) && parsableQuery.indexOf('"') >= 0); + if (bothModes) { + validateOneMode(parsableQuery, squareBracketQuotes, false); + validateOneMode(parsableQuery, squareBracketQuotes, true); + } else { + validateOneMode(parsableQuery, squareBracketQuotes, backslashEscapes); + } } - private static void validate(String query, boolean squareBracketQuotes, boolean backslashEscapes) { + private static void rejectLexerMismatchConstructs(String query) { + if (DOLLAR_QUOTE.matcher(query).find()) { + throw new IllegalArgumentException( + "Dollar-quoted string literals ($$...$$ or $tag$...$tag$) are not allowed in the query."); + } + if (ORACLE_QUOTE.matcher(query).find()) { + throw new IllegalArgumentException("Alternative-quoted string literals (q'...') are not allowed in the query."); + } + if (EXECUTABLE_COMMENT.matcher(query).find()) { + throw new IllegalArgumentException("Executable SQL comments ('/*!' or '/*M!') are not allowed in the query."); + } + } + + private static void validateOneMode(String query, boolean squareBracketQuotes, boolean backslashEscapes) { + lexicalCheck(query, squareBracketQuotes, backslashEscapes); + // Best-effort parser confirmation: if JSqlParser can parse the query, make sure it is a single SELECT. If it + // cannot parse it (a valid but dialect-specific construct), the lexical check above is authoritative. 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 + statements = CCJSqlParserUtil.parseStatements(query, PARSER_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(); + return; } - - // the parser returns null for an empty query - List nonEmptyStatements = statements == null ? Collections.emptyList() + // Statement separation is authoritative in the lexical scan above (it honors dialect comments such as MySQL + // '#', which JSqlParser does not), so the parser is only used here to confirm the statement type. + List nonEmpty = statements == null ? java.util.Collections.emptyList() : statements.stream().filter(Objects::nonNull).collect(Collectors.toList()); - if (nonEmptyStatements.size() != 1) { + if (nonEmpty.size() == 1 && !(nonEmpty.get(0) instanceof Select)) { throw new IllegalArgumentException(String.format( - "Only a single SELECT statement is allowed, but found %d statements.", nonEmptyStatements.size())); + "Only a SELECT statement is allowed. Found a %s statement.", nonEmpty.get(0).getClass().getSimpleName())); + } + } + + /** + * Scans the query, ignoring the contents of string literals, comments and quoted identifiers, and enforces that it + * is a single statement that reads and performs no write, schema change or locking. + */ + private static void lexicalCheck(String query, boolean squareBracketQuotes, boolean backslashEscapes) { + StringBuilder bareText = new StringBuilder(query.length()); + int statementsWithContent = 0; + boolean currentStatementHasContent = false; + int i = 0; + int n = query.length(); + while (i < n) { + char c = query.charAt(i); + if (c == '-' && i + 1 < n && query.charAt(i + 1) == '-') { + i = skipToEndOfLine(query, i + 2); + // A comment is a token boundary: emit a space so adjacent words do not glue into one token. + bareText.append(' '); + } else if (backslashEscapes && c == '#') { + // '#' starts a line comment in MySQL/MariaDB. Gated to backslash dialects so that in standard dialects, + // where '#' can be an operator, a '#' never hides a following statement from the scan. + i = skipToEndOfLine(query, i + 1); + bareText.append(' '); + } else if (c == '/' && i + 1 < n && query.charAt(i + 1) == '*') { + i = skipBlockComment(query, i); + bareText.append(' '); + } else if (c == '\'') { + i = skipString(query, i, '\'', backslashEscapes); + currentStatementHasContent = true; + // Replace the literal/identifier with a placeholder token surrounded by spaces, so neighbouring keywords + // stay separate tokens and a stripped identifier before '(' cannot make a preceding keyword look like a call. + bareText.append(" _ "); + } else if (c == '"') { + // A double quote may be a string (backslash can escape it) or an identifier; honor the dialect's backslash + // rule so a backslash-escaped quote cannot hide the real end of the token. + i = skipString(query, i, '"', backslashEscapes); + currentStatementHasContent = true; + bareText.append(" _ "); + } else if (c == '`') { + i = skipQuotedIdentifier(query, i, '`'); + currentStatementHasContent = true; + bareText.append(" _ "); + } else if (squareBracketQuotes && c == '[') { + i = skipBracketIdentifier(query, i); + currentStatementHasContent = true; + bareText.append(" _ "); + } else if (c == ';') { + if (currentStatementHasContent) { + statementsWithContent++; + } + currentStatementHasContent = false; + bareText.append(' '); + i++; + } else { + if (!Character.isWhitespace(c)) { + currentStatementHasContent = true; + } + bareText.append(c); + i++; + } } - Statement statement = nonEmptyStatements.get(0); - if (!(statement instanceof Select)) { + if (currentStatementHasContent) { + statementsWithContent++; + } + if (statementsWithContent > 1) { throw new IllegalArgumentException(String.format( - "Only a SELECT statement is allowed, but found a %s statement.", statement.getClass().getSimpleName())); + "Only a single SELECT statement is allowed. Found %d statements.", statementsWithContent)); } - validateSelect((Select) statement); + checkKeywords(bareText.toString()); } - private static void validateSelect(Select select) { - if (select.getWithItemsList() != null) { - for (WithItem withItem : select.getWithItemsList()) { - validateSelect(withItem.getSelect()); + private static void checkKeywords(String bareText) { + List words = new java.util.ArrayList<>(); + List endOffsets = new java.util.ArrayList<>(); + int i = 0; + int n = bareText.length(); + while (i < n) { + char c = bareText.charAt(i); + if (isIdentifierChar(c)) { + int start = i; + while (i < n && isIdentifierChar(bareText.charAt(i))) { + i++; + } + words.add(bareText.substring(start, i).toUpperCase()); + endOffsets.add(i); + } else { + i++; } } - 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."); + for (int w = 0; w < words.size(); w++) { + String word = words.get(w); + if (WRITE_KEYWORDS.contains(word)) { + boolean isAllowedFunctionCall = FUNCTION_COLLIDING_WRITE_KEYWORDS.contains(word) + && w > 0 + && isFollowedByOpenParen(bareText, endOffsets.get(w)); + if (!isAllowedFunctionCall) { + throw new IllegalArgumentException(String.format( + "The '%s' keyword is not allowed. The query must only read data.", word)); + } } - if (plainSelect.getForMode() != null) { - throw new IllegalArgumentException(String.format( - "SELECT ... FOR %s is not allowed, the query must only read data.", plainSelect.getForMode().getValue())); + if ("FOR".equals(word) && w + 1 < words.size() && LOCKING_AFTER_FOR.contains(words.get(w + 1))) { + throw new IllegalArgumentException( + "FOR UPDATE/SHARE locking clauses are not allowed. The query must only read data."); } - } else if (select instanceof SetOperationList) { - for (Select operand : ((SetOperationList) select).getSelects()) { - validateSelect(operand); + } + if (!words.contains("SELECT")) { + throw new IllegalArgumentException("Only a SELECT statement is allowed. No SELECT keyword was found."); + } + } + + private static boolean isFollowedByOpenParen(String text, int fromOffset) { + int i = fromOffset; + while (i < text.length() && Character.isWhitespace(text.charAt(i))) { + i++; + } + return i < text.length() && text.charAt(i) == '('; + } + + private static boolean isIdentifierChar(char c) { + return Character.isLetterOrDigit(c) || c == '_' || c == '$' || c >= 0x80; + } + + private static int skipToEndOfLine(String q, int i) { + while (i < q.length() && q.charAt(i) != '\n') { + i++; + } + return i; + } + + private static int skipBlockComment(String q, int i) { + i += 2; + while (i + 1 < q.length() && !(q.charAt(i) == '*' && q.charAt(i + 1) == '/')) { + i++; + } + return Math.min(q.length(), i + 2); + } + + private static int skipString(String q, int i, char quote, boolean backslashEscapes) { + i++; + while (i < q.length()) { + char c = q.charAt(i); + if (backslashEscapes && c == '\\' && i + 1 < q.length()) { + i += 2; + continue; } - } else if (select instanceof ParenthesedSelect) { - validateSelect(((ParenthesedSelect) select).getSelect()); - } else { - throw new IllegalArgumentException(String.format( - "Unsupported SELECT statement of type %s.", select.getClass().getSimpleName())); + if (c == quote) { + if (i + 1 < q.length() && q.charAt(i + 1) == quote) { + i += 2; + continue; + } + return i + 1; + } + i++; + } + throw new IllegalArgumentException("Unterminated string literal in the query."); + } + + private static int skipQuotedIdentifier(String q, int i, char quote) { + i++; + while (i < q.length()) { + char c = q.charAt(i); + if (c == quote) { + if (i + 1 < q.length() && q.charAt(i + 1) == quote) { + i += 2; + continue; + } + return i + 1; + } + i++; + } + throw new IllegalArgumentException("Unterminated quoted identifier in the query."); + } + + private static int skipBracketIdentifier(String q, int i) { + i++; + while (i < q.length()) { + char c = q.charAt(i); + if (c == ']') { + if (i + 1 < q.length() && q.charAt(i + 1) == ']') { + i += 2; + continue; + } + return i + 1; + } + i++; + } + throw new IllegalArgumentException("Unterminated quoted identifier in the query."); + } + + /** + * Validates that the given value is a single, bare column reference (optionally qualified and/or quoted) and + * nothing more. Used for inputs such as the split-by field, which are concatenated into generated SQL at runtime + * and so must not be able to carry operators, function calls, comments or additional statements. + * + * @param value the value to validate + * @param label a human readable name for the value, used in error messages + * @param squareBracketQuotes whether square brackets quote identifiers, as in SQL Server + * @throws IllegalArgumentException if the value is not a single column reference + */ + public static void validateColumnReference(String value, String label, boolean squareBracketQuotes) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(label + " must not be empty."); + } + // A column reference is concatenated into generated SQL, so reject anything that could end the identifier or the + // clause: statement separators, comments, backslashes and the lexer-mismatch quote constructs. + if (value.indexOf(';') >= 0 || value.indexOf('\\') >= 0 || value.indexOf('#') >= 0 + || value.contains("--") || value.contains("/*") + || DOLLAR_QUOTE.matcher(value).find() || ORACLE_QUOTE.matcher(value).find() + || EXECUTABLE_COMMENT.matcher(value).find()) { + throw new IllegalArgumentException(label + " must be a single column name."); + } + Expression expression; + try { + // allowPartialParse=false so trailing content (operators, ';', extra tokens) is rejected, not ignored + expression = CCJSqlParserUtil.parseExpression(value, false, + parser -> parser.withSquareBracketQuotation(squareBracketQuotes).withTimeOut(PARSE_TIMEOUT_MS)); + } catch (JSQLParserException e) { + throw new IllegalArgumentException(label + " must be a single column name. Error: " + firstLine(e), e); + } + if (!(expression instanceof Column)) { + throw new IllegalArgumentException(label + " must be a single column name."); } } diff --git a/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java b/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java index 6e04ede7c..9cd1f08c3 100644 --- a/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java +++ b/database-commons/src/test/java/io/cdap/plugin/util/SelectQueryValidatorTest.java @@ -25,7 +25,7 @@ public class SelectQueryValidatorTest { @Test - public void testValidSelectQueries() { + public void validate_singleReadOnlySelect_isAllowed() { String[] queries = { "SELECT * FROM my_table", "select id, name from my_table where id > 10 order by name", @@ -41,10 +41,23 @@ public void testValidSelectQueries() { "WITH t AS (SELECT id FROM my_table) SELECT * FROM t", "SELECT * FROM my_table WHERE name = 'a;b' -- trailing ; comment", "SELECT * FROM my_table /* ; DELETE FROM t */", + "SELECT * FROM my_table WHERE name = 'O;Brien -- not a comment'", "SELECT * FROM my_table;", "SELECT TOP 10 * FROM dbo.my_table", "SELECT * FROM my_table WHERE ROWNUM <= 10", "SELECT * FROM my_table LIMIT 10", + // functions whose names collide with write keywords must still be allowed + "SELECT TRUNCATE(col, 2) FROM my_table", + "SELECT INSERT(name, 1, 2, 'x') FROM my_table", + "SELECT REPLACE(name, 'a', 'b') FROM my_table", + // valid but dialect-specific syntax JSqlParser 4.9 cannot parse must still be allowed (lexical fallback) + "SELECT * FROM t TABLESAMPLE BERNOULLI (100.0 * 10 / (SELECT COUNT(*) FROM t))", + "SELECT * FROM t FOR READ ONLY", + "SELECT * FROM t FOR FETCH ONLY", + "LOCKING ROW FOR ACCESS SELECT * FROM t", + "SELECT * FROM `my``db`.`my``table`", + // a macro placeholder uses $ but is not a dollar-quote opener; must not be falsely rejected + "SELECT * FROM my_table WHERE ts <= '${logicalStartTime(yyyy-MM-dd,1d)}' AND $CONDITIONS", }; for (String query : queries) { try { @@ -56,73 +69,154 @@ public void testValidSelectQueries() { } @Test - public void testSquareBracketIdentifiers() { + public void validate_squareBracketIdentifiers_allowedButStackedRejected() { SelectQueryValidator.validate("SELECT * FROM [dbo].[my;table] WHERE [id] > 1", true); - try { - SelectQueryValidator.validate("SELECT * FROM [dbo].[t]; DELETE FROM [dbo].[t]", true); - Assert.fail("Stacked statements with square bracket identifiers should have been rejected"); - } catch (IllegalArgumentException e) { - // expected + assertRejected("SELECT * FROM [dbo].[t]; DELETE FROM [dbo].[t]", true); + } + + @Test + public void validate_backslashEscapedQuote_handledPerDialect() { + // MySQL/Databricks: a backslash escapes the quote, so this is a valid single query + SelectQueryValidator.validate("SELECT * FROM users WHERE last_name = 'O\\'Brien' AND id > 1", false, Boolean.TRUE); + // unknown dialect: the same text is ambiguous and must be rejected (it could hide a second statement) + assertRejected("SELECT 'a\\'; DELETE FROM my_table; --'", false); + } + + @Test + public void validate_lexerMismatchConstructs_areRejected() { + // PostgreSQL dollar-quoted strings (tagged and untagged) that hide a second statement + assertRejected("SELECT $$ ' $$; DELETE FROM my_table; -- '", false); + assertRejected("SELECT $tag$ ' $tag$; UPDATE my_table SET v=$$x$$ WHERE id=1; SELECT * FROM t WHERE 1=1 -- '", + false); + assertRejected("SELECT $$x$$ AS c FROM my_table", false); + // Oracle alternative-quoted literal + assertRejected("SELECT q'['; DELETE FROM my_table; --]' FROM dual", false); + // MySQL/MariaDB executable comments + assertRejected("SELECT * FROM my_table /*!50000 INTO OUTFILE '/tmp/out.csv' */", false); + assertRejected("SELECT * FROM my_table /*!50000 FOR UPDATE */", false); + assertRejected("SELECT 1; /*! DELETE FROM my_table */", false); + assertRejected("SELECT * FROM my_table /*M!50000 FOR UPDATE */", false); + } + + @Test + public void validate_stackedOrWriteStatements_areRejected() { + assertRejected("SELECT 1; DELETE FROM t", false); + assertRejected("SELECT * FROM t; UPDATE t SET v = 1", false); + assertRejected("SELECT * FROM \"pg_catalog\".\"pg_class\" LIMIT 0; UPDATE t SET v = 'x'; SELECT 1", false); + assertRejected("UPDATE my_table SET v = 1", false); + assertRejected("DELETE FROM my_table", false); + assertRejected("INSERT INTO my_table VALUES (1)", false); + assertRejected("DROP TABLE my_table", false); + assertRejected("TRUNCATE TABLE my_table", false); + assertRejected("CREATE TABLE t (id INT)", false); + assertRejected("ALTER TABLE my_table ADD COLUMN c INT", false); + assertRejected("MERGE INTO t USING s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.v = s.v", false); + assertRejected("GRANT SELECT ON my_table TO someone", false); + assertRejected("EXEC my_procedure", false); + assertRejected("CALL my_procedure()", false); + assertRejected("VALUES (1)", false); + assertRejected("TABLE my_table", false); + assertRejected("", false); + assertRejected(";", false); + } + + @Test + public void validate_intoOrForClauseIncludingSubqueries_areRejected() { + assertRejected("SELECT * INTO new_table FROM my_table", false); + assertRejected("SELECT * FROM my_table INTO OUTFILE '/tmp/out.csv'", false); + assertRejected("SELECT * FROM my_table FOR UPDATE", false); + assertRejected("SELECT * FROM my_table FOR SHARE", false); + // nested in subqueries (regression for subquery traversal gap) + assertRejected("SELECT * FROM (SELECT * INTO new_table FROM my_table) t", false); + assertRejected("SELECT * FROM (SELECT * FROM my_table FOR UPDATE) t", false); + assertRejected("SELECT * FROM my_table WHERE id IN (SELECT id FROM other_table FOR UPDATE)", false); + assertRejected("SELECT id FROM a UNION ALL (SELECT id INTO t2 FROM b)", false); + assertRejected("WITH d AS (DELETE FROM my_table RETURNING *) SELECT * FROM d", false); + // FOR READ ONLY / FOR FETCH ONLY are read-only and must be allowed + SelectQueryValidator.validate("SELECT * FROM my_table FOR READ ONLY", false); + SelectQueryValidator.validate("SELECT * FROM my_table FOR FETCH ONLY", false); + } + + @Test + public void validate_keywordsHiddenByCommentsOrQuotes_areRejected() { + // comments and quoted tokens must not glue adjacent words into one token and hide a keyword + assertRejected("SELECT * INTO/**/new_table FROM my_table", false); + assertRejected("SELECT * FROM my_table INTO/**/OUTFILE '/tmp/out.csv'", false); + assertRejected("SELECT * FROM my_table FOR/**/UPDATE", false); + assertRejected("SELECT * FROM \"my_table\"FOR\"x\"UPDATE", false); + assertRejected("SELECT a\"alias\"INTO new_table FROM my_table", false); + assertRejected("WITH d AS (DELETE/**/FROM my_table RETURNING *) SELECT * FROM d", false); + // a stripped identifier before '(' must not make INTO look like a function call + assertRejected("SELECT * INTO \"new_table\"(col1) FROM my_table", false); + // parenthesized write / dynamic SQL that JSqlParser cannot parse must not slip through the function exemption + assertRejected("UPDATE (SELECT * FROM my_table) SET v = 1", false); + assertRejected("EXEC (SELECT 'DROP TABLE x')", false); + assertRejected("EXECUTE (SELECT 'DROP TABLE x')", false); + } + + @Test + public void validate_doubleQuoteBackslashBreakout_isRejected() { + // In MySQL/Databricks a double quote is a string where a backslash can escape it; the query must be rejected + // whether or not backslash escaping is in effect, so it cannot break out under either interpretation. + String q = "SELECT \"a\\\"\"; DELETE FROM my_table; --\" FROM my_table"; + assertRejected(q, false); + Assert.assertThrows(IllegalArgumentException.class, () -> SelectQueryValidator.validate(q, false, Boolean.TRUE)); + } + + @Test + public void validate_nullQuery_isRejected() { + assertRejected(null, false); + } + + @Test + public void validate_invalidQueries_reportExpectedErrorMessages() { + assertError("SELECT 1; DELETE FROM t", "Only a single SELECT statement is allowed. Found 2 statements."); + assertError("DELETE FROM t", "The 'DELETE' keyword is not allowed. The query must only read data."); + assertError("SELECT * INTO t2 FROM t", "The 'INTO' keyword is not allowed. The query must only read data."); + assertError("SELECT * FROM t FOR UPDATE", + "FOR UPDATE/SHARE locking clauses are not allowed. The query must only read data."); + assertError("TABLE my_table", "Only a SELECT statement is allowed. No SELECT keyword was found."); + } + + @Test + public void validateColumnReference_singleColumnName_isAllowed() { + String[] columns = {"id", "my_col", "t.id", "\"My Col\"", "`id`", "schema.t.id"}; + for (String col : columns) { + SelectQueryValidator.validateColumnReference(col, "Split-By Field Name", false); } + SelectQueryValidator.validateColumnReference("[my id]", "Split-By Field Name", true); } @Test - public void testInvalidQueries() { - String[] queries = { - // stacked statements, as in the reported vulnerability - "SELECT * FROM \"pg_catalog\".\"pg_class\" LIMIT 0; UPDATE t SET v = 'x' WHERE id = 1; SELECT 1", - "SELECT * FROM my_table; DELETE FROM my_table", - "SELECT 1; SELECT 2", - // data modification and definition - "UPDATE my_table SET v = 1", - "DELETE FROM my_table", - "INSERT INTO my_table VALUES (1)", - "DROP TABLE my_table", - "TRUNCATE TABLE my_table", - "CREATE TABLE t (id INT)", - "ALTER TABLE my_table ADD COLUMN c INT", - "MERGE INTO t USING s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.v = s.v", - "GRANT SELECT ON my_table TO someone", - // session and database commands - "SET search_path TO other", - "USE other_db", - "ALTER SESSION SET NLS_DATE_FORMAT = 'YYYY-MM-DD'", - // stored procedures - "EXEC my_procedure", - "CALL my_procedure()", - // reads that write or lock - "SELECT * INTO new_table FROM my_table", - "SELECT * FROM my_table FOR UPDATE", - "WITH d AS (DELETE FROM my_table RETURNING *) SELECT * FROM d", - "SELECT * FROM my_table INTO OUTFILE '/tmp/out.csv'", - "SELECT * INTO OUTFILE '/tmp/out.csv' FROM my_table", - // a dollar quoted string (PostgreSQL) that hides a second statement - "SELECT $$ ' $$; DELETE FROM my_table; -- '", - // a string that hides a second statement when backslash escapes quotes (MySQL) but not otherwise - "SELECT 'a\\''; DELETE FROM my_table; -- '", - // a string that hides a second statement when backslash does not escape quotes (PostgreSQL) - "SELECT 'a\\'; DELETE FROM my_table; --'", - // empty and unparsable + public void validateColumnReference_nonColumnOrInjection_isRejected() { + String[] columns = { + "1 = 1); UPDATE audit_vault.write_marker SET value='INJECTED' WHERE id=1; SELECT * FROM \"my_table\" WHERE (1", + "id; DROP TABLE t", + "id)", + "f(id)", + "id, name", + "id OR 1=1", + "id /*! FOR UPDATE */", + "\"a\\\"\"; DROP TABLE t; --\"", + "id -- comment", + "id /* comment */", + "id # comment", "", - ";", - "SELECT FROM WHERE", + null, }; - for (String query : queries) { + for (String col : columns) { try { - SelectQueryValidator.validate(query, false); - Assert.fail(String.format("Query '%s' should have been rejected", query)); + SelectQueryValidator.validateColumnReference(col, "Split-By Field Name", false); + Assert.fail(String.format("Column '%s' should have been rejected", col)); } catch (IllegalArgumentException e) { // expected } } } - @Test - public void testErrorMessages() { - assertError("SELECT 1; DELETE FROM t", "Only a single SELECT statement is allowed, but found 2 statements."); - assertError("DELETE FROM t", "Only a SELECT statement is allowed, but found a Delete statement."); - assertError("SELECT * INTO t2 FROM t", "SELECT ... INTO is not allowed, the query must only read data."); - assertError("SELECT * FROM t FOR UPDATE", "SELECT ... FOR UPDATE is not allowed, the query must only read data."); + private static void assertRejected(String query, boolean squareBracketQuotes) { + Assert.assertThrows(IllegalArgumentException.class, + () -> SelectQueryValidator.validate(query, squareBracketQuotes)); } private static void assertError(String query, String expectedMessage) { diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java index 5997cf627..e13369b85 100644 --- a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java +++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksConnector.java @@ -45,6 +45,9 @@ import java.sql.SQLException; import java.util.HashMap; import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; +import java.util.stream.Stream; /** * Databricks Database Connector that connects to Databricks database via JDBC. @@ -115,16 +118,10 @@ protected SchemaReader getSchemaReader(String sessionID) { @Override protected String getTableName(String database, String schema, String table) { - if (database == null && schema == null) { - return quoteIdentifier(table, '`'); - } - if (database == null) { - return quoteIdentifier(schema, '`') + "." + quoteIdentifier(table, '`'); - } - if (schema == null) { - return quoteIdentifier(database, '`') + "." + quoteIdentifier(table, '`'); - } - return quoteIdentifier(database, '`') + "." + quoteIdentifier(schema, '`') + "." + quoteIdentifier(table, '`'); + return Stream.of(database, schema, table) + .filter(Objects::nonNull) + .map(part -> quoteIdentifier(part, '`')) + .collect(Collectors.joining(".")); } @Override diff --git a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSource.java b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSource.java index 524dec5ee..f2edcc5f4 100644 --- a/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSource.java +++ b/databricks-plugin/src/main/java/io/cdap/plugin/databricks/DatabricksSource.java @@ -78,6 +78,12 @@ protected String createConnectionString() { return connection == null ? null : connection.getConnectionString(); } + @Override + protected Boolean useBackslashEscapes() { + // In MySQL/MariaDB/SingleStore/Databricks a backslash escapes a quote inside a string literal. + return Boolean.TRUE; + } + @Override protected LineageRecorder getLineageRecorder(BatchSourceContext context) { DatabricksConnectorConfig connection = databricksSourceConfig.getConnection(); diff --git a/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java index 27b830c76..d97ec4e38 100644 --- a/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java +++ b/databricks-plugin/src/test/java/io/cdap/plugin/databricks/DatabricksConnectorUnitTest.java @@ -49,8 +49,8 @@ public void testGetTableNameEscapesBackticks() { @Test public void testGetStratifiedQueryEscapesStrata() { String query = CONNECTOR.getStratifiedQuery("`main`.`default`.`my_table`", 10, "id`; DROP TABLE t; --", "s1"); - Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`, RAND()")); - Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`\n")); + Assert.assertTrue(query, query.contains("ORDER BY `id``; DROP TABLE t; --`, RAND()")); + Assert.assertTrue(query, query.contains("ORDER BY `id``; DROP TABLE t; --`\n")); } @Test diff --git a/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java b/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java index f11342461..ed2aeec9c 100644 --- a/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java +++ b/generic-database-plugin/src/test/java/io/cdap/plugin/db/sink/DBSourceTestRun.java @@ -383,6 +383,56 @@ public void testNonSelectImportQueryMacroFailsRun() throws Exception { Assert.assertEquals(rowCount, countRows("my_table")); } + @Test + public void testNonSelectBoundingQueryFailsRun() throws Exception { + // The bounding query is only used at runtime, so a non-SELECT bounding query is rejected in prepareRun and the + // pipeline run fails (deployment succeeds). The injected DELETE must never run. + int rowCount = countRows("my_table"); + ETLPlugin sourceConfig = new ETLPlugin( + DatabaseConstants.PLUGIN_NAME, + BatchSource.PLUGIN_TYPE, + ImmutableMap.builder() + .put(ConnectionConfig.CONNECTION_STRING, getConnectionURL()) + .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME) + .put(AbstractDBSource.DBSourceConfig.IMPORT_QUERY, "SELECT * FROM \"my_table\" WHERE $CONDITIONS") + .put(AbstractDBSource.DBSourceConfig.BOUNDING_QUERY, + "SELECT MIN(ID), MAX(ID) FROM \"my_table\"; DELETE FROM \"my_table\"") + .put(AbstractDBSource.DBSourceConfig.SPLIT_BY, "ID") + .put(AbstractDBSource.DBSourceConfig.NUM_SPLITS, "2") + .put(Constants.Reference.REFERENCE_NAME, "DBNonSelectBoundingQueryTest") + .build(), + null); + ApplicationManager appManager = deployETL(sourceConfig, MockSink.getPlugin("nonSelectBoundingOutput"), + DATAPIPELINE_ARTIFACT, "dbSourceNonSelectBoundingQueryTest"); + WorkflowManager workflowManager = appManager.getWorkflowManager(SmartWorkflow.NAME); + workflowManager.start(); + workflowManager.waitForRun(ProgramRunStatus.FAILED, 5, TimeUnit.MINUTES); + Assert.assertEquals(rowCount, countRows("my_table")); + } + + @Test + public void testInjectionSplitByFailsDeployment() throws Exception { + int rowCount = countRows("my_table"); + String splitBy = "ID); UPDATE \"my_table\" SET NAME='INJECTED' WHERE (1=1"; + ETLPlugin sourceConfig = new ETLPlugin( + DatabaseConstants.PLUGIN_NAME, + BatchSource.PLUGIN_TYPE, + ImmutableMap.builder() + .put(ConnectionConfig.CONNECTION_STRING, getConnectionURL()) + .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME) + .put(AbstractDBSource.DBSourceConfig.IMPORT_QUERY, "SELECT * FROM \"my_table\" WHERE $CONDITIONS") + .put(AbstractDBSource.DBSourceConfig.BOUNDING_QUERY, "SELECT MIN(ID),MAX(ID) FROM \"my_table\"") + .put(AbstractDBSource.DBSourceConfig.SPLIT_BY, splitBy) + .put(AbstractDBSource.DBSourceConfig.NUM_SPLITS, "2") + .put(Constants.Reference.REFERENCE_NAME, "DBInjectionSplitByTest") + .build(), + null); + ETLBatchConfig etlConfig = getETLBatchConfig(sourceConfig, MockSink.getPlugin("injectionSplitByOutput")); + assertDeploymentFailure(NamespaceId.DEFAULT.app("dbSourceInjectionSplitByTest"), etlConfig, + DATAPIPELINE_ARTIFACT, "Deploying DB Source with an injection in split-by must fail."); + Assert.assertEquals(rowCount, countRows("my_table")); + } + private int countRows(String table) throws SQLException { try (Connection connection = getConnection(); Statement statement = connection.createStatement(); diff --git a/mariadb-plugin/src/main/java/io/cdap/plugin/mariadb/MariadbSource.java b/mariadb-plugin/src/main/java/io/cdap/plugin/mariadb/MariadbSource.java index 6361ba86c..a82813dd7 100644 --- a/mariadb-plugin/src/main/java/io/cdap/plugin/mariadb/MariadbSource.java +++ b/mariadb-plugin/src/main/java/io/cdap/plugin/mariadb/MariadbSource.java @@ -64,6 +64,12 @@ protected String createConnectionString() { mariadbSourceConfig.host, mariadbSourceConfig.port, mariadbSourceConfig.database); } + @Override + protected Boolean useBackslashEscapes() { + // In MySQL/MariaDB/SingleStore/Databricks a backslash escapes a quote inside a string literal. + return Boolean.TRUE; + } + @Override protected Class getDBRecordType() { return MariadbDBRecord.class; diff --git a/memsql-plugin/src/main/java/io/cdap/plugin/memsql/source/MemsqlSource.java b/memsql-plugin/src/main/java/io/cdap/plugin/memsql/source/MemsqlSource.java index b9d003441..1f887826c 100644 --- a/memsql-plugin/src/main/java/io/cdap/plugin/memsql/source/MemsqlSource.java +++ b/memsql-plugin/src/main/java/io/cdap/plugin/memsql/source/MemsqlSource.java @@ -50,4 +50,10 @@ protected String createConnectionString() { return String.format(MemsqlConstants.MEMSQL_CONNECTION_STRING_FORMAT, memsqlSourceConfig.host, memsqlSourceConfig.port, memsqlSourceConfig.database); } + + @Override + protected Boolean useBackslashEscapes() { + // In MySQL/MariaDB/SingleStore/Databricks a backslash escapes a quote inside a string literal. + return Boolean.TRUE; + } } diff --git a/mssql-plugin/src/e2e-test/resources/pluginParameters.properties b/mssql-plugin/src/e2e-test/resources/pluginParameters.properties index 4421c0c72..92e1a5604 100644 --- a/mssql-plugin/src/e2e-test/resources/pluginParameters.properties +++ b/mssql-plugin/src/e2e-test/resources/pluginParameters.properties @@ -89,3 +89,4 @@ outputDatatypesSchema4=[{"key":"ID","value":"string"},{"key":"COL1","value":"dat {"key":"COL2","value":"datetime"},{"key":"COL3","value":"datetime"}, {"key":"COL4","value":"timestamp"}] mssqlOutputDatatypesSchema=[{"key":"Description","value":"string"}, {"key":"FileData","value":"bytes"}, \ {"key":"Notes","value":"string"}, {"key":"CustomValue","value":"string"}] +invalidBoundingQueryValue=SELECT MIN(id), MAX(id) FROM mytable; DELETE FROM mytable diff --git a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerAction.java b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerAction.java index 62c7b9252..c067c46a1 100644 --- a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerAction.java +++ b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerAction.java @@ -125,7 +125,8 @@ public Map getDBSpecificArguments() { @Override public List getInitQueries() { if (!Strings.isNullOrEmpty(currentLanguage)) { - return Collections.singletonList(String.format(SqlServerConstants.SET_LANGUAGE_QUERY_FORMAT, currentLanguage)); + return Collections.singletonList( + String.format(SqlServerConstants.SET_LANGUAGE_QUERY_FORMAT, currentLanguage.replace("'", "''"))); } return Collections.emptyList(); diff --git a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerPostAction.java b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerPostAction.java index 6540223ed..3c3d83343 100644 --- a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerPostAction.java +++ b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerPostAction.java @@ -125,7 +125,8 @@ public Map getDBSpecificArguments() { @Override public List getInitQueries() { if (!Strings.isNullOrEmpty(currentLanguage)) { - return Collections.singletonList(String.format(SqlServerConstants.SET_LANGUAGE_QUERY_FORMAT, currentLanguage)); + return Collections.singletonList( + String.format(SqlServerConstants.SET_LANGUAGE_QUERY_FORMAT, currentLanguage.replace("'", "''"))); } return Collections.emptyList(); diff --git a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSink.java b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSink.java index 00b627139..e365ffbfc 100644 --- a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSink.java +++ b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSink.java @@ -206,7 +206,8 @@ protected SqlServerConnectorConfig getConnection() { @Override public List getInitQueries() { if (!Strings.isNullOrEmpty(currentLanguage)) { - return Collections.singletonList(String.format(SqlServerConstants.SET_LANGUAGE_QUERY_FORMAT, currentLanguage)); + return Collections.singletonList( + String.format(SqlServerConstants.SET_LANGUAGE_QUERY_FORMAT, currentLanguage.replace("'", "''"))); } return Collections.emptyList(); diff --git a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java index b6bcb9ff9..2e1f4e044 100644 --- a/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java +++ b/mssql-plugin/src/main/java/io/cdap/plugin/mssql/SqlServerSource.java @@ -201,7 +201,8 @@ protected AbstractDBSpecificConnectorConfig getConnection() { @Override public List getInitQueries() { if (!Strings.isNullOrEmpty(currentLanguage)) { - return Collections.singletonList(String.format(SqlServerConstants.SET_LANGUAGE_QUERY_FORMAT, currentLanguage)); + return Collections.singletonList( + String.format(SqlServerConstants.SET_LANGUAGE_QUERY_FORMAT, currentLanguage.replace("'", "''"))); } return Collections.emptyList(); diff --git a/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java b/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java index 231d072b9..493e2f0e5 100644 --- a/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java +++ b/mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerConnectorUnitTest.java @@ -51,8 +51,8 @@ public void getTableNameEscapesQuotesTest() { @Test public void getStratifiedQueryEscapesStrataTest() { String query = CONNECTOR.getStratifiedQuery("\"db\".\"dbo\".\"table\"", 10, "id\"; DROP TABLE t; --", "s1"); - Assert.assertTrue(query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\", RAND()")); - Assert.assertTrue(query.endsWith("ORDER BY \"id\"\"; DROP TABLE t; --\"")); + Assert.assertTrue(query, query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\", RAND()")); + Assert.assertTrue(query, query.endsWith("ORDER BY \"id\"\"; DROP TABLE t; --\"")); } /** diff --git a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlAction.java b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlAction.java index c8433b92e..bd471f367 100644 --- a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlAction.java +++ b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlAction.java @@ -115,7 +115,8 @@ public List getInitQueries() { initQueries.add(MysqlConstants.ANSI_QUOTES_QUERY); } if (!Strings.isNullOrEmpty(sqlMode)) { - initQueries.add(String.format(MysqlConstants.SET_SQL_MODE_QUERY_FORMAT, sqlMode)); + initQueries.add( + String.format(MysqlConstants.SET_SQL_MODE_QUERY_FORMAT, sqlMode.replace("\\", "\\\\").replace("'", "''"))); } return initQueries; } diff --git a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlPostAction.java b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlPostAction.java index 6ae62a3cf..6d3f820c2 100644 --- a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlPostAction.java +++ b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlPostAction.java @@ -115,7 +115,8 @@ public List getInitQueries() { initQueries.add(MysqlConstants.ANSI_QUOTES_QUERY); } if (!Strings.isNullOrEmpty(sqlMode)) { - initQueries.add(String.format(MysqlConstants.SET_SQL_MODE_QUERY_FORMAT, sqlMode)); + initQueries.add( + String.format(MysqlConstants.SET_SQL_MODE_QUERY_FORMAT, sqlMode.replace("\\", "\\\\").replace("'", "''"))); } return initQueries; } diff --git a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSink.java b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSink.java index 6487ffae6..335d962d4 100644 --- a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSink.java +++ b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSink.java @@ -217,7 +217,8 @@ public void validate(FailureCollector collector) { @Override public List getInitQueries() { if (!Strings.isNullOrEmpty(sqlMode)) { - return Collections.singletonList(String.format(MysqlConstants.SET_SQL_MODE_QUERY_FORMAT, sqlMode)); + return Collections.singletonList( + String.format(MysqlConstants.SET_SQL_MODE_QUERY_FORMAT, sqlMode.replace("\\", "\\\\").replace("'", "''"))); } return Collections.emptyList(); } diff --git a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSource.java b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSource.java index d7d7c400b..d88b8200a 100644 --- a/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSource.java +++ b/mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSource.java @@ -65,6 +65,12 @@ protected String createConnectionString() { return mysqlSourceConfig.getConnectionString(); } + @Override + protected Boolean useBackslashEscapes() { + // In MySQL/MariaDB/SingleStore/Databricks a backslash escapes a quote inside a string literal. + return Boolean.TRUE; + } + @Override protected Class getDBRecordType() { return MysqlDBRecord.class; @@ -191,7 +197,8 @@ public List getInitQueries() { initQueries.add(MysqlConstants.ANSI_QUOTES_QUERY); } if (!Strings.isNullOrEmpty(sqlMode)) { - initQueries.add(String.format(MysqlConstants.SET_SQL_MODE_QUERY_FORMAT, sqlMode)); + initQueries.add( + String.format(MysqlConstants.SET_SQL_MODE_QUERY_FORMAT, sqlMode.replace("\\", "\\\\").replace("'", "''"))); } return initQueries; } diff --git a/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java b/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java index 01820842e..d9d0d1717 100644 --- a/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java +++ b/mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlConnectorUnitTest.java @@ -57,8 +57,8 @@ public void getTableNameEscapesBackticksTest() { @Test public void getStratifiedQueryEscapesStrataTest() { String query = CONNECTOR.getStratifiedQuery("`db`.`table`", 10, "id`; DROP TABLE t; --", "s1"); - Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`, RAND()")); - Assert.assertTrue(query.contains("ORDER BY `id``; DROP TABLE t; --`\n")); + Assert.assertTrue(query, query.contains("ORDER BY `id``; DROP TABLE t; --`, RAND()")); + Assert.assertTrue(query, query.contains("ORDER BY `id``; DROP TABLE t; --`\n")); } /** diff --git a/oracle-plugin/src/e2e-test/resources/errorMessage.properties b/oracle-plugin/src/e2e-test/resources/errorMessage.properties index 10d147773..a366fbe20 100644 --- a/oracle-plugin/src/e2e-test/resources/errorMessage.properties +++ b/oracle-plugin/src/e2e-test/resources/errorMessage.properties @@ -14,8 +14,7 @@ errorMessageBlankUsername=Username is required when password is given. errorMessageInvalidTableName=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '08006', errorCode: '17002', errorMessage: SQL Exception occurred: [Message='IO Error: Unknown host specified ', SQLState='08006', ErrorCode='17002'].' errorMessageInvalidSinkDatabase=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '08006', errorCode: '17002' errorMessageInvalidHost=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '08006', errorCode: '17002', errorMessage: SQL Exception occurred: [Message='IO Error: Unknown host specified ', SQLState='08006', ErrorCode='17002'].' -errorLogsMessageInvalidBoundingQuery=Spark program 'phase-1' failed with error: Stage 'Oracle' encountered : \ - java.io.IOException: ORA-00936: missing expression . Please check the system logs for more details. +errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid Bounding Query blank.database.message=Required property 'database' has no value. blank.connection.message=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '72000', errorCode: '1005', errorMessage: SQL Exception occurred: [Message='ORA-01005: null password given; logon denied ', SQLState='72000', ErrorCode='1005'].' blank.HostBlank.message=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '08006', errorCode: '17002', errorMessage: SQL Exception occurred: [Message='IO Error: The Network Adapter could not establish the connection', SQLState='08006', ErrorCode='17002'].' diff --git a/oracle-plugin/src/e2e-test/resources/pluginParameters.properties b/oracle-plugin/src/e2e-test/resources/pluginParameters.properties index 266418425..cde56ee18 100644 --- a/oracle-plugin/src/e2e-test/resources/pluginParameters.properties +++ b/oracle-plugin/src/e2e-test/resources/pluginParameters.properties @@ -84,7 +84,7 @@ invalidPort=testPort 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 invalidOracleTableKey=asdas diff --git a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java index 2441862d0..c73eea17f 100644 --- a/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java +++ b/oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleConnectorUnitTest.java @@ -52,8 +52,8 @@ public void getTableNameEscapesQuotesTest() { @Test public void getStratifiedQueryEscapesStrataTest() { String query = CONNECTOR.getStratifiedQuery("\"SCHEMA\".\"TABLE\"", 10, "ID\", 1) --", "s1"); - Assert.assertTrue(query.contains("ORDER BY \"ID\"\", 1) --\", dbms_random.random")); - Assert.assertTrue(query.contains("ORDER BY \"ID\"\", 1) --\" )")); + Assert.assertTrue(query, query.contains("ORDER BY \"ID\"\", 1) --\", dbms_random.random")); + Assert.assertTrue(query, query.contains("ORDER BY \"ID\"\", 1) --\" )")); } /** diff --git a/postgresql-plugin/src/e2e-test/resources/errorMessage.properties b/postgresql-plugin/src/e2e-test/resources/errorMessage.properties index 9dacc0b67..4160e9e60 100644 --- a/postgresql-plugin/src/e2e-test/resources/errorMessage.properties +++ b/postgresql-plugin/src/e2e-test/resources/errorMessage.properties @@ -18,4 +18,4 @@ errorMessageInvalidSourceHost=SQL error while getting query schema: The connecti 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 errorMessageInvalidHost=Error encountered while configuring the stage: 'SQL Error occurred, sqlState: '08001', errorCode: '0', errorMessage: SQL Exception occurred: [Message='The connection attempt failed.', SQLState='08001', ErrorCode='0'].' -errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid +errorLogsMessageInvalidBoundingQuery=Errors were encountered during validation. Invalid Bounding Query diff --git a/postgresql-plugin/src/e2e-test/resources/pluginParameters.properties b/postgresql-plugin/src/e2e-test/resources/pluginParameters.properties index bad2510be..8dd9fbe7b 100644 --- a/postgresql-plugin/src/e2e-test/resources/pluginParameters.properties +++ b/postgresql-plugin/src/e2e-test/resources/pluginParameters.properties @@ -71,7 +71,7 @@ invalidPort=testPort 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 #POSTGRESQL Valid Properties PostgreSQLConnectionArgumentsList=fetchsize=1000 diff --git a/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java b/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java index 57e6399d4..47c5f5e94 100644 --- a/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java +++ b/postgresql-plugin/src/test/java/io/cdap/plugin/postgres/PostgresConnectorUnitTest.java @@ -59,8 +59,8 @@ public void getTableNameEscapesQuotesTest() { @Test public void getStratifiedQueryEscapesStrataTest() { String query = CONNECTOR.getStratifiedQuery("\"schema\".\"table\"", 10, "id\"; DROP TABLE t; --", "s1"); - Assert.assertTrue(query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\", random()")); - Assert.assertTrue(query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\"\n")); + Assert.assertTrue(query, query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\", random()")); + Assert.assertTrue(query, query.contains("ORDER BY \"id\"\"; DROP TABLE t; --\"\n")); } /**