[BUG-SEC-1] SQL Injection in DebeziumOffsetStorage — String.format constructs SQL with user-controlled topic names #1288

Summary

DebeziumOffsetStorage.java constructs SQL queries using String.format() with user-controlled Debezium topic names interpolated directly into the SQL string. This is a SQL injection vulnerability — a maliciously crafted topic name can execute arbitrary SQL against the ClickHouse offset storage table.

Affected Code

File: sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumOffsetStorage.java

// Line ~80: Query construction
String query = String.format("select * from %s.%s where topic_name='%s'", 
    database, tableName, topicName);

// Line ~100: Delete construction  
String deleteQuery = String.format("delete from %s.%s where topic_name='%s'",
    database, tableName, topicName);

The topicName parameter comes from Debezium's internal topic naming, which incorporates the MySQL server name from the connector config. While the server name is typically controlled by the admin, this is still a SQL injection vector:

1. If an attacker can influence the server name in the config (e.g., via the REST API or environment variable injection), they can inject arbitrary SQL
2. The pattern violates defense-in-depth — all SQL should use parameterized queries
3. ClickHouse supports parameterized queries via PreparedStatement, which is already used elsewhere in the codebase

Fix

Replace String.format with a parameterized query:

String query = "SELECT * FROM " + database + "." + tableName + " WHERE topic_name = ?";
PreparedStatement ps = conn.prepareStatement(query);
ps.setString(1, topicName);
ResultSet rs = ps.executeQuery();

(Database and table names should also be validated/escaped, but at minimum the WHERE clause value must be parameterized.)

Impact

• Severity: HIGH
• Type: SQL Injection (CWE-89)
• Affected versions: All versions with DebeziumOffsetStorage

Reproduction

Set database.server.name to a value containing a single quote followed by SQL, e.g.:
test' OR 1=1; --

The resulting query becomes:

select * from offset_db.offset_table where topic_name='test' OR 1=1; --'