From 7226732f8852c62c5b741bc997975a18c27f105e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 09:08:02 +0900 Subject: [PATCH 1/4] test(cdc): require safe processed-data replica diagnostics --- ...rocessedDataReplicaApplierLoggingTest.java | 127 ++++++++++++++++++ 1 file changed, 127 insertions(+) create mode 100644 cdc-service/src/test/java/com/xtrmetl/cdc/replication/ProcessedDataReplicaApplierLoggingTest.java diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/replication/ProcessedDataReplicaApplierLoggingTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/ProcessedDataReplicaApplierLoggingTest.java new file mode 100644 index 00000000..dbe58d24 --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/replication/ProcessedDataReplicaApplierLoggingTest.java @@ -0,0 +1,127 @@ +package com.xtrmetl.cdc.replication; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.slf4j.LoggerFactory; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +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.ArgumentMatchers.startsWith; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; + +/** + * Verifies that row-replication observability keeps business identifiers and parser diagnostics out of logs. + */ +@ExtendWith(OutputCaptureExtension.class) +class ProcessedDataReplicaApplierLoggingTest { + + @Test + void missingDataFailureDoesNotExposeRowIdentifier(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + ProcessedDataReplicaApplier applier = applier(jdbcTemplate); + long sensitiveRowId = 918273645L; + String topic = "xtrmetl-cdc.public.processed_data"; + String keyJson = "{\"payload\":{\"id\":" + sensitiveRowId + "}}"; + String valueJson = "{\"payload\":{\"op\":\"u\",\"after\":{\"id\":" + sensitiveRowId + "}}}"; + + IllegalStateException failure = assertThrows( + IllegalStateException.class, + () -> applier.apply(topic, keyJson, valueJson) + ); + + assertEquals("Missing data field in CDC event", failure.getMessage()); + verifyNoInteractions(jdbcTemplate); + assertSafeLogs(output, "Replica apply failed: missing data field", Long.toString(sensitiveRowId)); + } + + @Test + void nullDataFailureDoesNotExposeRowIdentifier(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + ProcessedDataReplicaApplier applier = applier(jdbcTemplate); + long sensitiveRowId = 817263544L; + String topic = "xtrmetl-cdc.public.processed_data"; + String keyJson = "{\"payload\":{\"id\":" + sensitiveRowId + "}}"; + String valueJson = "{\"payload\":{\"op\":\"u\",\"after\":{\"id\":" + sensitiveRowId + ",\"data\":null}}}"; + + IllegalStateException failure = assertThrows( + IllegalStateException.class, + () -> applier.apply(topic, keyJson, valueJson) + ); + + assertEquals("CDC event data is null", failure.getMessage()); + verifyNoInteractions(jdbcTemplate); + assertSafeLogs(output, "Replica apply failed: data is null", Long.toString(sensitiveRowId)); + } + + @Test + void malformedValueLogsOnlyStableClassification(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + ProcessedDataReplicaApplier applier = applier(jdbcTemplate); + String malformedValue = "{buyer-contract-secret-8472"; + + applier.apply("xtrmetl-cdc.public.processed_data", null, malformedValue); + + verifyNoInteractions(jdbcTemplate); + assertSafeLogs( + output, + "Failed to parse Debezium value JSON; skipping replica apply", + "buyer-contract-secret-8472", + "JsonParseException", + "ReaderBasedJsonParser" + ); + } + + @Test + void malformedKeyFallbackLogsOnlyStableClassification(CapturedOutput output) { + JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class); + ProcessedDataReplicaApplier applier = applier(jdbcTemplate); + String malformedKey = "{buyer-key-secret-8472"; + String valueJson = "{\"payload\":{\"op\":\"c\",\"after\":{\"id\":7,\"data\":\"accepted\"}}}"; + Logger logger = (Logger) LoggerFactory.getLogger(ProcessedDataReplicaApplier.class); + Level previousLevel = logger.getLevel(); + logger.setLevel(Level.DEBUG); + try { + applier.apply("xtrmetl-cdc.public.processed_data", malformedKey, valueJson); + } finally { + logger.setLevel(previousLevel); + } + + verify(jdbcTemplate).update(startsWith("INSERT INTO processed_data"), eq(7L), eq("accepted")); + assertSafeLogs( + output, + "Failed to parse Debezium key JSON; falling back to value payload", + "buyer-key-secret-8472", + "JsonParseException", + "ReaderBasedJsonParser" + ); + } + + private static ProcessedDataReplicaApplier applier(JdbcTemplate jdbcTemplate) { + return new ProcessedDataReplicaApplier( + jdbcTemplate, + new ObjectMapper(), + Set.of("processed_data") + ); + } + + 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); + } + } +} From cd21cb36405ee2f3c2cf88ba223223bf00580afb Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 09:10:56 +0900 Subject: [PATCH 2/4] fix(cdc): minimize row replica diagnostics --- .../ProcessedDataReplicaApplier.java | 26 ++++++++++++++----- 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/cdc-service/src/main/java/com/xtrmetl/cdc/replication/ProcessedDataReplicaApplier.java b/cdc-service/src/main/java/com/xtrmetl/cdc/replication/ProcessedDataReplicaApplier.java index 1c0467f6..f776296a 100644 --- a/cdc-service/src/main/java/com/xtrmetl/cdc/replication/ProcessedDataReplicaApplier.java +++ b/cdc-service/src/main/java/com/xtrmetl/cdc/replication/ProcessedDataReplicaApplier.java @@ -69,6 +69,20 @@ public ProcessedDataReplicaApplier( this.sqlByTable = Map.copyOf(compiledSql); } + /** + * Applies one Debezium row event to an explicitly configured replica table. + * + *

Irrelevant topics, tombstones, malformed envelopes, and events without an identifier are + * ignored without exposing raw event or parser diagnostics. A create or update event whose + * {@code data} field is missing or null fails with a stable non-sensitive exception message. + * Delete and upsert identifiers and values remain JDBC-bound and are never interpolated into + * generated SQL.

+ * + * @param topic Kafka topic carrying the Debezium event + * @param keyJson optional Debezium key JSON used to locate the row identifier + * @param valueJson optional Debezium value JSON containing the change envelope + * @throws IllegalStateException when a create or update event lacks usable {@code data} + */ public void apply(@Nullable String topic, @Nullable String keyJson, @Nullable String valueJson) { if (topic == null) { return; @@ -104,14 +118,14 @@ public void apply(@Nullable String topic, @Nullable String keyJson, @Nullable St JsonNode after = envelope.after(); if (after == null || after.isNull() || !after.has("data")) { - log.error("Replica apply failed: missing data field (topic={}, id={})", topic, id); - throw new IllegalStateException("Missing data field in CDC event for id=" + id); + log.error("Replica apply failed: missing data field"); + throw new IllegalStateException("Missing data field in CDC event"); } JsonNode dataNode = after.get("data"); if (dataNode.isNull()) { - log.error("Replica apply failed: data is null (topic={}, id={})", topic, id); - throw new IllegalStateException("data is null in CDC event for id=" + id); + log.error("Replica apply failed: data is null"); + throw new IllegalStateException("CDC event data is null"); } String data = dataNode.isTextual() ? dataNode.asText() : dataNode.toString(); @@ -186,7 +200,7 @@ private DebeziumEnvelope parseDebeziumEnvelope(String valueJson) { return new DebeziumEnvelope(op, after); } catch (IOException e) { - log.warn("Failed to parse Debezium value JSON; skipping replica apply", e); + log.warn("Failed to parse Debezium value JSON; skipping replica apply"); return null; } } @@ -210,7 +224,7 @@ private Long extractIdFromKey(String keyJson) { JsonNode payload = root.has("payload") ? root.get("payload") : root; return extractLong(payload, "id"); } catch (IOException e) { - log.debug("Failed to parse Debezium key JSON; falling back to value payload", e); + log.debug("Failed to parse Debezium key JSON; falling back to value payload"); return null; } } From 8986887ca7d3802b02115feedf3425862009c755 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 09:14:26 +0900 Subject: [PATCH 3/4] test(docs): require CDC diagnostic confidentiality doctoring --- ...osticConfidentialityDocumentationTest.java | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) create mode 100644 cdc-service/src/test/java/com/xtrmetl/cdc/ops/CdcDiagnosticConfidentialityDocumentationTest.java diff --git a/cdc-service/src/test/java/com/xtrmetl/cdc/ops/CdcDiagnosticConfidentialityDocumentationTest.java b/cdc-service/src/test/java/com/xtrmetl/cdc/ops/CdcDiagnosticConfidentialityDocumentationTest.java new file mode 100644 index 00000000..7e8847fb --- /dev/null +++ b/cdc-service/src/test/java/com/xtrmetl/cdc/ops/CdcDiagnosticConfidentialityDocumentationTest.java @@ -0,0 +1,54 @@ +package com.xtrmetl.cdc.ops; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Keeps CDC diagnostic-confidentiality guidance and authoritative security references discoverable. + */ +class CdcDiagnosticConfidentialityDocumentationTest { + + @Test + void doctoringDocumentsPurposeBoundDiagnosticMinimizationAndPrimarySecurityReferences() throws IOException { + Path doctoring = findProjectRoot().resolve("docs/doctoring/cdc-diagnostic-confidentiality.md"); + assertTrue(Files.isRegularFile(doctoring), + "cdc diagnostic-confidentiality doctoring must be checked in"); + + String body = Files.readString(doctoring, StandardCharsets.UTF_8); + assertTrue(body.contains("CWE-532"), "doctoring must map the weakness to CWE-532"); + assertTrue(body.contains("OWASP Logging Cheat Sheet"), + "doctoring must cite current OWASP logging guidance"); + assertTrue(body.contains("row identifiers"), + "doctoring must explain why business row identifiers do not belong in ordinary logs"); + assertTrue(body.contains("parser") && body.contains("exception"), + "doctoring must cover parser exception diagnostics"); + assertTrue(body.contains("purpose-bound") || body.contains("purpose limitation"), + "doctoring must distinguish minimization from blanket masking"); + assertTrue(body.contains("APA 7"), "doctoring must identify the reference style"); + } + + private static Path findProjectRoot() { + Path current = Paths.get(System.getProperty("user.dir")).toAbsolutePath(); + Path lastPom = null; + while (current != null) { + if (Files.exists(current.resolve(".git"))) { + return current; + } + if (Files.exists(current.resolve("pom.xml"))) { + lastPom = current; + } + current = current.getParent(); + } + if (lastPom != null) { + return lastPom; + } + throw new IllegalStateException("project root not found"); + } +} From e2be226f1844fa9980b7311aa70dba15041d0c87 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 10 Aug 2026 09:18:25 +0900 Subject: [PATCH 4/4] docs(security): document CDC diagnostic confidentiality --- .../cdc-diagnostic-confidentiality.md | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) create mode 100644 docs/doctoring/cdc-diagnostic-confidentiality.md diff --git a/docs/doctoring/cdc-diagnostic-confidentiality.md b/docs/doctoring/cdc-diagnostic-confidentiality.md new file mode 100644 index 00000000..e65f232a --- /dev/null +++ b/docs/doctoring/cdc-diagnostic-confidentiality.md @@ -0,0 +1,54 @@ +# CDC diagnostic confidentiality + +## Status and scope + +This doctoring note defines the evidence and design boundary for CDC diagnostics that originate from untrusted connector, database, parser, or replicated business data. It is not a claim that every related active pull request is shipped. Protected `develop` remains authoritative for implemented behavior; active repairs such as #170, #171, #172, and #174 remain unshipped until protected integration. + +The control is **purpose-bound diagnostic minimization**, not blanket masking of business data. CDC computation may require row identifiers, DDL, connector metadata, or raw event values to perform its authorized task. Ordinary API responses, status payloads, and logs have a narrower operational purpose and therefore should carry finite outcome classifications rather than unnecessary raw values. + +## Threat boundary + +The following values are sensitive or high-risk diagnostic material unless a separately reviewed purpose requires them: + +- row identifiers and other high-cardinality business identifiers; +- JDBC/database connection strings, hosts, usernames, password-like parameters, and provider coordinates; +- raw DDL, SQL, schema/object names, literals, comments, storage paths, or tenant/customer labels; +- raw Debezium key/value JSON and replicated payload fragments; +- parser, driver, broker, database, filesystem, and connector exception messages or stack traces; +- access tokens, session identifiers, secrets, or credential-adjacent values. + +Length truncation is not a confidentiality control. A truncated connection string, DDL statement, row identifier, or parser exception can still disclose the sensitive value. + +## Required controls + +1. **Stable public/status failures.** Public API and health/status representations use bounded, non-sensitive error classifications and messages. They do not concatenate `Exception.getMessage()` or driver diagnostics. +2. **Finite ordinary logs.** Logs retain outcome, subsystem, and a bounded topic/service classification only when operationally necessary. Raw row identifiers, DDL, key/value JSON, SQL, connection strings, exception messages, and stack traces are excluded by default. +3. **Preserve in-process causality where needed.** Programmatic callers may retain the original exception or suppressed exception when the existing execution contract requires causal handling. The exception object does not need to be serialized into ordinary logs merely because it remains available in-process. +4. **No regex-only masking boundary.** Secret-pattern replacement is not the primary design because it cannot enumerate every sensitive identifier or provider diagnostic. Prefer not transporting the untrusted diagnostic in the first place. +5. **Purpose limitation.** Business values remain available to the authorized replication computation. Observability receives only the minimum information required for operation, incident classification, and safe support. +6. **Tests at the real boundary.** Log/API/status tests inject realistic sensitive diagnostics and prove both non-disclosure and preservation of functional behavior such as JDBC execution, fallback, lifecycle, or failure propagation. + +## Current traceability + +| Work item | Boundary | Canonical maturity | +|---|---|---| +| #170 | replication-slot status and probe diagnostics | `active_pr` | +| #171 | schema-change DDL and execution/parser diagnostics in logs | `active_pr` | +| #172 | CDC stop failure text returned by the API | `active_pr` | +| #173 / #174 | replicated row identifiers and parser diagnostics in `ProcessedDataReplicaApplier` | `active_pr` | + +These rows are dated implementation evidence, not a substitute for the canonical Security, Threat Model, Test Strategy, Operability, and Traceability graph maintained through #149/#159. + +## Security mapping + +MITRE **CWE-532** identifies insertion of sensitive information into a log file as a weakness. The OWASP Logging Cheat Sheet similarly cautions against directly recording sensitive personal data, passwords, access tokens, database connection strings, and other secrets. mightyETL applies those principles by separating computational authority from diagnostic transport instead of destructively masking data required for the authorized ETL/CDC workflow. + +## Acceptance implications + +A CDC diagnostic-confidentiality repair is incomplete when it merely shortens, partially masks, or relocates a raw diagnostic. Acceptance requires realistic RED evidence at the intended source boundary, the narrowest root-cause GREEN repair, focused and full behavioral verification, current security/dependency/SBOM evidence, and canonical documentation/traceability reconciliation. GitHub synthetic-merge execution is useful compatibility evidence but is not relabeled literal-source proof when governance requires the literal pull-request head. + +## References — APA 7 + +The MITRE Corporation. (2026). *CWE-532: Insertion of sensitive information into log file (Version 4.20).* Common Weakness Enumeration. https://cwe.mitre.org/data/definitions/532.html + +The OWASP Foundation. (n.d.). *OWASP Logging Cheat Sheet.* OWASP Cheat Sheet Series. Retrieved August 10, 2026, from https://cheatsheetseries.owasp.org/cheatsheets/Logging_Cheat_Sheet.html