Skip to content

fix(cdc): rebuild DDL log confidentiality on live develop #267

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
seonghobae wants to merge 4 commits into from
Closed

fix(cdc): rebuild DDL log confidentiality on live develop #267

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9c724e3
test(cdc): prove replicated DDL stays out of logs
seonghobae Aug 11, 2026
017d135
fix(cdc): keep DDL diagnostics out of logs
seonghobae Aug 11, 2026
2d6a651
test(cdc): cover blocked DDL log confidentiality
seonghobae Aug 11, 2026
45e1409
test(cdc): exercise retained NUL log guard
seonghobae Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 12 additions & 27 deletions cdc-service/src/main/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,14 +24,14 @@
*
* <p>DDL execution is disabled by default. When enabled, the default policy is a positive
* allow-list, only a single comment-free statement is accepted, and prefix matches must end
* at a SQL token boundary.
* at a SQL token boundary. Ordinary lifecycle logs retain bounded outcome metadata without
* reproducing raw DDL statements or provider exception diagnostics.</p>
*/
@Service
@ConditionalOnProperty(prefix = "xtrmetl.replica", name = "enabled", havingValue = "true")
public class SchemaChangeReplicaApplier {

private static final Logger log = LoggerFactory.getLogger(SchemaChangeReplicaApplier.class);
private static final int DDL_LOG_MAX_LENGTH = 500;
private static final Set<String> IDEMPOTENT_DDL_SQL_STATES = Set.of(
"42P06",
"42P07",
Expand Down Expand Up @@ -94,20 +94,17 @@ public void apply(@Nullable String topic, @Nullable String keyJson, @Nullable St
// single-statement, comment-free, and configured policy gates above.
jdbcTemplate.execute(ddl); // nosemgrep: java.spring.security.audit.spring-sqli.spring-sqli
if (log.isInfoEnabled()) {
log.info("Applied schema change DDL on replica (topic={}, ddl={})",
topic, truncateForLog(ddl));
log.info("Applied schema change DDL on replica (topic={})", topic);
}
} catch (DataAccessException e) {
if (isIdempotentDuplicate(e)) {
if (log.isInfoEnabled()) {
log.info("Schema change DDL already applied; skipping duplicate (topic={}, ddl={})",
topic, truncateForLog(ddl));
log.info("Schema change DDL already applied; skipping duplicate (topic={})", topic);
}
return;
}
if (log.isErrorEnabled()) {
log.error("Failed to apply schema change DDL on replica (topic={}, ddl={})",
topic, truncateForLog(ddl), e);
log.error("Failed to apply schema change DDL on replica (topic={})", topic);
}
throw e;
}
Expand All @@ -119,15 +116,15 @@ private String requireSingleStatement(String topic, String ddl) {
trimmed = trimmed.substring(0, trimmed.length() - 1).trim();
}
if (trimmed.contains(";")) {
logBlocked(topic, ddl, "Blocked multi-statement DDL");
logBlocked(topic, "Blocked multi-statement DDL");
throw new IllegalArgumentException("Multiple SQL statements are not allowed");
}
return trimmed;
}

private String requireCommentFree(String topic, String ddl) {
if (ddl.contains("--") || ddl.contains("/*") || ddl.contains("*/") || ddl.indexOf('\0') >= 0) {
logBlocked(topic, ddl, "Blocked DDL containing SQL comments or NUL");
logBlocked(topic, "Blocked DDL containing SQL comments or NUL");
throw new IllegalArgumentException("SQL comments and NUL characters are not allowed in replicated DDL");
}
return ddl;
Expand All @@ -142,7 +139,7 @@ private String extractDdl(String valueJson) {
}
return payload.path("ddl").asText(null);
} catch (IOException e) {
log.warn("Failed to parse Debezium schema change JSON; skipping DDL apply", e);
log.warn("Failed to parse Debezium schema change JSON; skipping DDL apply");
return null;
}
}
Expand Down Expand Up @@ -230,21 +227,20 @@ private void validateDdl(String topic, String ddl) {
String normalized = normalizeForValidation(ddl);
if (ddlValidationMode == DdlValidationMode.BLOCKLIST
&& ddlBlockedPrefixes.stream().anyMatch(prefix -> matchesPrefix(normalized, prefix))) {
logBlocked(topic, ddl, "Blocked DDL by validation policy");
logBlocked(topic, "Blocked DDL by validation policy");
throw new IllegalArgumentException("DDL blocked by validation policy");
}

if (ddlValidationMode == DdlValidationMode.WHITELIST
&& ddlAllowedPrefixes.stream().noneMatch(prefix -> matchesPrefix(normalized, prefix))) {
logBlocked(topic, ddl, "Blocked DDL by validation policy");
logBlocked(topic, "Blocked DDL by validation policy");
throw new IllegalArgumentException("DDL blocked by validation policy");
}
}

private void logBlocked(String topic, String ddl, String message) {
private void logBlocked(String topic, String message) {
if (log.isWarnEnabled()) {
log.warn("{} (mode={}, topic={}, ddl={})",
message, ddlValidationMode, topic, truncateForLog(ddl));
log.warn("{} (mode={}, topic={})", message, ddlValidationMode, topic);
}
}

Expand All @@ -268,17 +264,6 @@ private static String normalizeForValidation(String ddl) {
return ddl.trim().replaceAll("\\s+", " ").toUpperCase(Locale.ROOT);
}

private static String truncateForLog(String ddl) {
if (ddl == null) {
return null;
}
String normalized = ddl.trim().replaceAll("\\s+", " ");
if (normalized.length() <= DDL_LOG_MAX_LENGTH) {
return normalized;
}
return normalized.substring(0, DDL_LOG_MAX_LENGTH) + "...";
}

private enum DdlValidationMode {
NONE,
WHITELIST,
Expand Down
222 changes: 222 additions & 0 deletions ...vice/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,222 @@
package com.xtrmetl.cdc.replication;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.boot.test.system.CapturedOutput;
import org.springframework.boot.test.system.OutputCaptureExtension;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;

import java.sql.SQLException;
import java.util.Map;

import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;

/**
* Verifies that schema-replication observability never republishes raw DDL or driver diagnostics.
*/
@ExtendWith(OutputCaptureExtension.class)
class SchemaChangeReplicaApplierLoggingTest {

@Test
void successfulApplyLogsOnlyBoundedMetadataNotRawDdl(CapturedOutput output) {
JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
String secretLiteral = "buyer-contract-secret-8472";
String inputDdl = "CREATE TABLE confidential_record(secret_value text DEFAULT '" + secretLiteral + "')";
String executedDdl = "CREATE TABLE IF NOT EXISTS confidential_record(secret_value text DEFAULT '"
+ secretLiteral + "')";

applier.apply(schemaTopic(), null, ddlEnvelope(inputDdl));

verify(jdbcTemplate).execute(eq(executedDdl));
assertSafeLogs(output, "Applied schema change DDL on replica", secretLiteral, "DEFAULT", "ddl=");
}

@Test
void blockedDdlLogsPolicyOutcomeWithoutRawStatement(CapturedOutput output) {
JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
SchemaChangeReplicaApplier applier = applier(
jdbcTemplate,
"whitelist",
"CREATE TABLE,ALTER TABLE,CREATE INDEX",
""
);
String secretPath = "/srv/private/buyer-contract-secret-8472";
String ddl = "CREATE TABLESPACE reporting LOCATION '" + secretPath + "'";

assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, ddlEnvelope(ddl)));

verifyNoInteractions(jdbcTemplate);
assertSafeLogs(output, "Blocked DDL by validation policy", secretPath, "TABLESPACE", "ddl=");
}

@Test
void multiStatementBlockLogsClassificationWithoutRawStatement(CapturedOutput output) {
JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
String secretLiteral = "buyer-contract-secret-8472";
String ddl = "CREATE TABLE confidential_record(id int); DROP TABLE " + secretLiteral;

assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, ddlEnvelope(ddl)));

verifyNoInteractions(jdbcTemplate);
assertSafeLogs(output, "Blocked multi-statement DDL", secretLiteral, "DROP TABLE", "ddl=");
}

@Test
void sqlCommentBlockLogsClassificationWithoutRawStatement(CapturedOutput output) {
JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
String secretLiteral = "buyer-contract-secret-8472";
String ddl = "CREATE TABLE confidential_record(id int) -- " + secretLiteral;

assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, ddlEnvelope(ddl)));

verifyNoInteractions(jdbcTemplate);
assertSafeLogs(
output,
"Blocked DDL containing SQL comments or NUL",
secretLiteral,
"confidential_record",
"ddl="
);
}

@Test
void nulBlockLogsClassificationWithoutRawStatement(CapturedOutput output) throws Exception {
JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
String secretLiteral = "buyer-contract-secret-8472";
String ddl = "CREATE TABLE confidential_record(id int) " + (char) 0 + secretLiteral;
String envelope = new ObjectMapper().writeValueAsString(Map.of("payload", Map.of("ddl", ddl)));

assertThrows(IllegalArgumentException.class, () -> applier.apply(schemaTopic(), null, envelope));

verifyNoInteractions(jdbcTemplate);
assertSafeLogs(
output,
"Blocked DDL containing SQL comments or NUL",
secretLiteral,
"confidential_record",
"ddl="
);
}

@Test
void duplicateDdlLogsOutcomeWithoutSqlOrDriverDiagnostics(CapturedOutput output) {
JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
String secretLiteral = "buyer-contract-secret-8472";
String inputDdl = "CREATE TABLE confidential_record(secret_value text DEFAULT '" + secretLiteral + "')";
String executedDdl = "CREATE TABLE IF NOT EXISTS confidential_record(secret_value text DEFAULT '"
+ secretLiteral + "')";
SQLException sqlException = new SQLException(
"relation already exists at jdbc:postgresql://db.internal/prod?password=driver-secret",
"42P07"
);
DataAccessException duplicate = new DataAccessException("driver-secret", sqlException) {};
doThrow(duplicate).when(jdbcTemplate).execute(eq(executedDdl));

assertDoesNotThrow(() -> applier.apply(schemaTopic(), null, ddlEnvelope(inputDdl)));

verify(jdbcTemplate).execute(eq(executedDdl));
assertSafeLogs(
output,
"Schema change DDL already applied; skipping duplicate",
secretLiteral,
"driver-secret",
"jdbc:postgresql://",
"ddl="
);
}

@Test
void executionFailureLogsOutcomeWithoutSqlOrDriverDiagnostics(CapturedOutput output) {
JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
String secretLiteral = "buyer-contract-secret-8472";
String inputDdl = "CREATE TABLE confidential_record(secret_value text DEFAULT '" + secretLiteral + "')";
String executedDdl = "CREATE TABLE IF NOT EXISTS confidential_record(secret_value text DEFAULT '"
+ secretLiteral + "')";
DataAccessException failure = new DataAccessException(
"jdbc:postgresql://db.internal/prod?password=driver-secret"
) {};
doThrow(failure).when(jdbcTemplate).execute(eq(executedDdl));

DataAccessException thrown = assertThrows(
DataAccessException.class,
() -> applier.apply(schemaTopic(), null, ddlEnvelope(inputDdl))
);

assertSame(failure, thrown);
assertSafeLogs(
output,
"Failed to apply schema change DDL on replica",
secretLiteral,
"driver-secret",
"jdbc:postgresql://",
"ddl="
);
}

@Test
void malformedEventLogsOnlyStableParseClassification(CapturedOutput output) {
JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
SchemaChangeReplicaApplier applier = applier(jdbcTemplate, "none", "", "");
String malformed = "{buyer-contract-secret-8472";

assertDoesNotThrow(() -> applier.apply(schemaTopic(), null, malformed));

verifyNoInteractions(jdbcTemplate);
assertSafeLogs(
output,
"Failed to parse Debezium schema change JSON; skipping DDL apply",
"buyer-contract-secret-8472",
"JsonParseException",
"ReaderBasedJsonParser"
);
}

private static SchemaChangeReplicaApplier applier(
JdbcTemplate jdbcTemplate,
String validationMode,
String allowedPrefixes,
String blockedPrefixes
) {
return new SchemaChangeReplicaApplier(
jdbcTemplate,
new ObjectMapper(),
true,
validationMode,
allowedPrefixes,
blockedPrefixes
);
}

private static String schemaTopic() {
return "xtrmetl-cdc.schema-changes";
}

private static String ddlEnvelope(String ddl) {
return "{\"payload\":{\"ddl\":\"" + ddl.replace("'", "\\u0027") + "\"}}";
}

private static void assertSafeLogs(CapturedOutput output, String expected, String... forbidden) {
String logs = output.getOut() + output.getErr();
assertTrue(logs.contains(expected));
for (String value : forbidden) {
assertFalse(logs.contains(value), () -> "Log output exposed forbidden value: " + value);
}
}
}
Loading