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.
*/ @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 SetEvents outside the schema-change topic family, blank payloads, and envelopes without DDL + * are ignored. Malformed JSON is classified as an unavailable event and skipped without + * publishing parser diagnostics. Policy violations fail closed with an exception before JDBC + * execution. Non-duplicate JDBC failures are rethrown after a bounded diagnostic event.
+ * + * @param topic source Kafka topic; only the schema-change suffix is accepted + * @param keyJson optional Debezium key, currently unused by schema application + * @param valueJson Debezium value envelope containing a {@code ddl} field + */ public void apply(@Nullable String topic, @Nullable String keyJson, @Nullable String valueJson) { if (!ddlEnabled || topic == null @@ -94,20 +116,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; } @@ -119,7 +138,7 @@ 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; @@ -127,12 +146,13 @@ private String requireSingleStatement(String topic, String ddl) { 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; } + @Nullable private String extractDdl(String valueJson) { try { JsonNode root = objectMapper.readTree(valueJson); @@ -142,7 +162,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; } } @@ -230,21 +250,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); } } @@ -268,17 +287,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, diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java new file mode 100644 index 00000000..0c65f67b --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java @@ -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); + } + } +}