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. Raw DDL and database-driver diagnostics are deliberately excluded + * from ordinary logs because statements can contain deployment-sensitive literals or names.
*/ @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 @@ -84,55 +106,47 @@ public void apply(@Nullable String topic, @Nullable String keyJson, @Nullable St return; } - ddl = requireSingleStatement(topic, ddl); - ddl = requireCommentFree(topic, ddl); + ddl = requireSingleStatement(ddl); + ddl = requireCommentFree(ddl); ddl = makeIdempotent(ddl); - validateDdl(topic, ddl); + validateDdl(ddl); try { // DDL identifiers cannot be JDBC bind parameters. The statement has passed the // 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"); } 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"); 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"); throw e; } } - private String requireSingleStatement(String topic, String ddl) { + private String requireSingleStatement(String ddl) { String trimmed = ddl.trim(); while (trimmed.endsWith(";")) { trimmed = trimmed.substring(0, trimmed.length() - 1).trim(); } if (trimmed.contains(";")) { - logBlocked(topic, ddl, "Blocked multi-statement DDL"); + logBlocked("Blocked multi-statement DDL"); throw new IllegalArgumentException("Multiple SQL statements are not allowed"); } return trimmed; } - private String requireCommentFree(String topic, String ddl) { + private String requireCommentFree(String ddl) { if (ddl.contains("--") || ddl.contains("/*") || ddl.contains("*/") || ddl.indexOf('\0') >= 0) { - logBlocked(topic, ddl, "Blocked DDL containing SQL comments or NUL"); + logBlocked("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 +156,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; } } @@ -222,7 +236,7 @@ private boolean isIdempotentDuplicate(DataAccessException exception) { return false; } - private void validateDdl(String topic, String ddl) { + private void validateDdl(String ddl) { if (ddlValidationMode == DdlValidationMode.NONE) { return; } @@ -230,22 +244,19 @@ 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("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("Blocked DDL by validation policy"); throw new IllegalArgumentException("DDL blocked by validation policy"); } } - private void logBlocked(String topic, String ddl, String message) { - if (log.isWarnEnabled()) { - log.warn("{} (mode={}, topic={}, ddl={})", - message, ddlValidationMode, topic, truncateForLog(ddl)); - } + private void logBlocked(String message) { + log.warn("{} (mode={})", message, ddlValidationMode); } private static boolean matchesPrefix(String normalizedDdl, String normalizedPrefix) { @@ -268,17 +279,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..f68f2468 --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/SchemaChangeReplicaApplierLoggingTest.java @@ -0,0 +1,169 @@ +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 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 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); + } + } +}