> legacyDone = new ConcurrentHashMap<>();
+
private Database database;
private String mapTable;
private String sequenceTable;
@@ -90,64 +101,120 @@ public void open() throws HopException {
database.setAutoCommit(false);
}
+ @Override
+ public String findOrCreate(String patternName, String sourceKey, MaskAllocator allocator)
+ throws HopException {
+ return findOrCreate(patternName, sourceKey, null, allocator);
+ }
+
+ /**
+ * Finds or stores the replacement in one transaction. When another process stores the same source
+ * key, the next sequence row or the same replacement at the same moment, the constraint violation
+ * rolls the attempt back and the next attempt sees what that process committed.
+ *
+ * A row under the legacy key moves to {@code sourceKey}. When {@code sourceKey} already has a
+ * row, for example because a second spelling folds onto the same key, the legacy row is removed
+ * so the original value does not stay in the table.
+ */
@Override
public synchronized String findOrCreate(
- String patternName, String sourceKey, MaskAllocator allocator) throws HopException {
+ String patternName, String sourceKey, KeySupplier legacyKey, MaskAllocator allocator)
+ throws HopException {
Map patternCache =
cache.computeIfAbsent(patternName, k -> new ConcurrentHashMap<>());
+ Set patternLegacyDone =
+ legacyDone.computeIfAbsent(patternName, k -> ConcurrentHashMap.newKeySet());
+ String oldKey = legacyKey == null ? null : legacyKey.get();
+ if (oldKey != null && (oldKey.equals(sourceKey) || patternLegacyDone.contains(oldKey))) {
+ oldKey = null;
+ }
String cached = patternCache.get(sourceKey);
- if (cached != null) {
+ if (cached != null && oldKey == null) {
return cached;
}
ensureOpen();
- try {
- String existing = lookup(patternName, sourceKey);
- if (existing != null) {
+ for (int attempt = 1; ; attempt++) {
+ try {
+ String masked = cached != null ? cached : lookup(patternName, sourceKey);
+ if (oldKey != null) {
+ String oldMasked = lookup(patternName, oldKey);
+ if (oldMasked != null) {
+ if (masked == null) {
+ rekey(patternName, oldKey, sourceKey);
+ masked = oldMasked;
+ } else {
+ delete(patternName, oldKey);
+ }
+ }
+ }
+ if (masked == null) {
+ masked = allocator.allocate(this);
+ insertMap(patternName, sourceKey, masked);
+ }
database.commit();
- patternCache.put(sourceKey, existing);
- return existing;
- }
- String created = allocator.allocate(this);
- insertMap(patternName, sourceKey, created);
- database.commit();
- patternCache.put(sourceKey, created);
- return created;
- } catch (HopException e) {
- rollbackQuietly();
- if (isConstraintViolation(e)) {
- String winner = lookup(patternName, sourceKey);
- if (winner != null) {
- database.commit();
- patternCache.put(sourceKey, winner);
- return winner;
+ patternCache.put(sourceKey, masked);
+ if (oldKey != null) {
+ patternLegacyDone.add(oldKey);
+ }
+ return masked;
+ } catch (HopException e) {
+ rollbackQuietly();
+ if (!isConstraintViolation(e)) {
+ throw e;
+ }
+ if (attempt >= MAX_ATTEMPTS) {
+ // The database's message names the conflicting key, which can be the original value.
+ // Leave the cause out so that value does not reach the log.
+ throw new HopException(
+ "Unable to store a replacement for pattern '"
+ + patternName
+ + "' after "
+ + MAX_ATTEMPTS
+ + " attempts: other processes kept storing conflicting rows in "
+ + mapTable);
}
}
- throw e;
}
}
+ /**
+ * Next sequence value. The update locks the pattern's row until the caller commits, so a second
+ * process waits and then continues from the committed value.
+ */
@Override
public synchronized long allocateSequence(String patternName, long start) throws HopException {
ensureOpen();
- Long current = queryLong(selectNextSql(), oneString(), new Object[] {patternName});
- if (current == null) {
- exec(
- "INSERT INTO "
- + sequenceTable
- + " ("
- + quotedPattern
- + ", "
- + quotedNext
- + ") VALUES (?, ?)",
- stringAndLong(),
- new Object[] {patternName, start + 1});
- return start;
+ Result result =
+ database.execStatement(
+ "UPDATE "
+ + sequenceTable
+ + " SET "
+ + quotedNext
+ + " = "
+ + quotedNext
+ + " + 1 WHERE "
+ + quotedPattern
+ + " = ?",
+ oneString(),
+ new Object[] {patternName});
+ if (result.getNrLinesUpdated() > 0) {
+ Long next = queryLong(selectNextSql(), oneString(), new Object[] {patternName});
+ if (next == null) {
+ throw new HopException("The sequence row for pattern '" + patternName + "' disappeared");
+ }
+ return next - 1;
}
exec(
- "UPDATE " + sequenceTable + " SET " + quotedNext + " = ? WHERE " + quotedPattern + " = ?",
- longAndString(),
- new Object[] {current + 1, patternName});
- return current;
+ "INSERT INTO "
+ + sequenceTable
+ + " ("
+ + quotedPattern
+ + ", "
+ + quotedNext
+ + ") VALUES (?, ?)",
+ stringAndLong(),
+ new Object[] {patternName, start + 1});
+ return start;
}
@Override
@@ -164,6 +231,7 @@ public synchronized int count(String patternName) throws HopException {
@Override
public synchronized void close() {
cache.clear();
+ legacyDone.clear();
if (database != null) {
database.disconnect();
database = null;
@@ -175,7 +243,9 @@ private void createMapTable() throws HopException {
return;
}
// Primary-key columns have to be NOT NULL. The generic CREATE TABLE leaves them nullable, and
- // H2 rejects a primary key on a nullable column.
+ // H2 rejects a primary key on a nullable column. The unique key stops two source values from
+ // sharing a replacement, which keeps the masked value short enough to index on every database.
+ // Tables created by earlier versions keep their layout.
database.execStatement(
"CREATE TABLE "
+ mapTable
@@ -184,11 +254,15 @@ private void createMapTable() throws HopException {
+ ", "
+ columnDefinition(stringColumn(COLUMN_SOURCE, 255))
+ ", "
- + columnDefinition(stringColumn(COLUMN_MASKED, 2000))
+ + columnDefinition(stringColumn(COLUMN_MASKED, MASKED_LENGTH))
+ ", PRIMARY KEY ("
+ quotedPattern
+ ", "
+ quotedSource
+ + "), UNIQUE ("
+ + quotedPattern
+ + ", "
+ + quotedMasked
+ "))");
}
@@ -249,6 +323,28 @@ private void insertMap(String patternName, String sourceKey, String masked) thro
new Object[] {patternName, sourceKey, masked});
}
+ private void rekey(String patternName, String oldKey, String newKey) throws HopException {
+ exec(
+ "UPDATE "
+ + mapTable
+ + " SET "
+ + quotedSource
+ + " = ? WHERE "
+ + quotedPattern
+ + " = ? AND "
+ + quotedSource
+ + " = ?",
+ rekeyParameters(),
+ new Object[] {newKey, patternName, oldKey});
+ }
+
+ private void delete(String patternName, String sourceKey) throws HopException {
+ exec(
+ "DELETE FROM " + mapTable + " WHERE " + quotedPattern + " = ? AND " + quotedSource + " = ?",
+ twoStrings(),
+ new Object[] {patternName, sourceKey});
+ }
+
private String selectNextSql() {
return "SELECT " + quotedNext + " FROM " + sequenceTable + " WHERE " + quotedPattern + " = ?";
}
@@ -342,16 +438,17 @@ private static RowMeta threeStrings() {
return meta;
}
- private static IRowMeta stringAndLong() {
- RowMeta meta = oneString();
- meta.addValueMeta(new ValueMetaInteger(COLUMN_NEXT));
+ private static RowMeta rekeyParameters() {
+ RowMeta meta = new RowMeta();
+ meta.addValueMeta(new ValueMetaString(COLUMN_SOURCE));
+ meta.addValueMeta(new ValueMetaString(COLUMN_PATTERN));
+ meta.addValueMeta(new ValueMetaString("old_" + COLUMN_SOURCE));
return meta;
}
- private static IRowMeta longAndString() {
- RowMeta meta = new RowMeta();
+ private static IRowMeta stringAndLong() {
+ RowMeta meta = oneString();
meta.addValueMeta(new ValueMetaInteger(COLUMN_NEXT));
- meta.addValueMeta(new ValueMetaString(COLUMN_PATTERN));
return meta;
}
}
diff --git a/plugins/transforms/maskfields/src/main/java/org/apache/hop/pipeline/transforms/maskfields/store/IMaskingStore.java b/plugins/transforms/maskfields/src/main/java/org/apache/hop/pipeline/transforms/maskfields/store/IMaskingStore.java
index cb0380c6efd..c2b90d44241 100644
--- a/plugins/transforms/maskfields/src/main/java/org/apache/hop/pipeline/transforms/maskfields/store/IMaskingStore.java
+++ b/plugins/transforms/maskfields/src/main/java/org/apache/hop/pipeline/transforms/maskfields/store/IMaskingStore.java
@@ -31,6 +31,18 @@ public interface IMaskingStore extends AutoCloseable {
String findOrCreate(String patternName, String sourceKey, MaskAllocator allocator)
throws HopException;
+ /**
+ * Like {@link #findOrCreate(String, String, MaskAllocator)}, but also finds a mapping stored
+ * under the key an earlier version wrote, and moves it to {@code sourceKey}, or removes it when
+ * {@code sourceKey} already has a mapping. A store that never kept mappings across versions
+ * ignores it.
+ */
+ default String findOrCreate(
+ String patternName, String sourceKey, KeySupplier legacyKey, MaskAllocator allocator)
+ throws HopException {
+ return findOrCreate(patternName, sourceKey, allocator);
+ }
+
/** Next sequence value for the pattern, starting at {@code start} the first time. */
long allocateSequence(String patternName, long start) throws HopException;
@@ -40,6 +52,12 @@ String findOrCreate(String patternName, String sourceKey, MaskAllocator allocato
@Override
void close();
+ /** Supplies a key only when the store needs it. */
+ @FunctionalInterface
+ interface KeySupplier {
+ String get() throws HopException;
+ }
+
/** Builds the replacement string for a new source key. */
@FunctionalInterface
interface MaskAllocator {
diff --git a/plugins/transforms/maskfields/src/main/resources/org/apache/hop/pipeline/transforms/maskfields/messages/messages_en_US.properties b/plugins/transforms/maskfields/src/main/resources/org/apache/hop/pipeline/transforms/maskfields/messages/messages_en_US.properties
index c2292826e69..98728217008 100644
--- a/plugins/transforms/maskfields/src/main/resources/org/apache/hop/pipeline/transforms/maskfields/messages/messages_en_US.properties
+++ b/plugins/transforms/maskfields/src/main/resources/org/apache/hop/pipeline/transforms/maskfields/messages/messages_en_US.properties
@@ -46,10 +46,13 @@ MaskFields.Check.Incompatible=Field ''{0}'' cannot use masking pattern ''{1}'':
MaskFields.Check.NoConnection=Masking pattern ''{0}'' stores mappings in a database and has no connection.
MaskFields.Check.MissingConnection=Database connection ''{0}'' for masking pattern ''{1}'' was not found.
MaskFields.Check.NoTable=Masking pattern ''{0}'' stores mappings in a database and has no table name.
+MaskFields.Check.MaskedTooLong=Masking pattern ''{0}'' can write values of up to {1} characters. A mapping table created by this version holds {2}.
+MaskFields.Check.PlainTextKeys=Masking pattern ''{0}'' has no hash secret, so its mapping table stores the original values in plain text.
MaskFields.Error.FieldMissing=Field ''{0}'' is no longer in the input row.
MaskFields.Error.Convert=Unable to write the masked value into field ''{0}'' with pattern ''{1}''.
MaskFields.Error.BadSequence=Sequence start ''{0}'' on masking pattern ''{1}'' is not an integer.
+MaskFields.Error.HashSecretUnresolved=The hash secret of masking pattern ''{0}'' is empty, holds a variable that is not set, or does not decrypt.
MaskFields.Error.OpenStore=Unable to open the mapping table for masking pattern ''{0}''.
MaskingRules.SetEmpty=Set to blank only applies to a string field.
@@ -89,10 +92,16 @@ MaskingPattern.Suffix.Tooltip=Text written after the token.
MaskingPattern.SequenceStart.Label=Sequence start
MaskingPattern.SequenceStart.Tooltip=First integer to use. The default is 1.
MaskingPattern.Storage.Label=Remember values
-MaskingPattern.Storage.Tooltip=NONE assigns a new value every row. MEMORY keeps the mapping for this run. DATABASE keeps it in a table across runs. The table stores the original value.
+MaskingPattern.Storage.Tooltip=NONE assigns a new value every row. MEMORY keeps the mapping for this run. DATABASE keeps it in a table across runs.
+MaskingPattern.TrimKey.Label=Ignore surrounding spaces
+MaskingPattern.TrimKey.Tooltip=Values that only differ in leading or trailing spaces get the same replacement. A value of only spaces is left unchanged.
+MaskingPattern.IgnoreCase.Label=Ignore case
+MaskingPattern.IgnoreCase.Tooltip=Values that only differ in upper and lower case get the same replacement.
+MaskingPattern.HashSecret.Label=Hash secret
+MaskingPattern.HashSecret.Tooltip=The mapping table stores an HMAC-SHA256 of each value made with this secret, instead of the value itself. Use a variable such as '${MASKING_SECRET}' and keep the secret out of the project. Changing the secret starts a new mapping. Without a secret the table stores the original values.
MaskingPattern.Connection.Label=Connection
MaskingPattern.Connection.Tooltip=Database that holds the mapping table.
MaskingPattern.Schema.Label=Schema
MaskingPattern.Schema.Tooltip=Schema of the mapping table. Leave this empty to use the connection default.
MaskingPattern.Table.Label=Table
-MaskingPattern.Table.Tooltip=Mapping table. A second table with the suffix _seq stores the next sequence number. Both are created when they are missing.
+MaskingPattern.Table.Tooltip=Mapping table. A second table with the suffix _seq stores the next sequence number. Both are created when they are missing. A new mapping table holds replacements of up to 255 characters.
diff --git a/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskFieldsMetaTest.java b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskFieldsMetaTest.java
index 8cd58245a98..130046b72e5 100644
--- a/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskFieldsMetaTest.java
+++ b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskFieldsMetaTest.java
@@ -18,6 +18,7 @@
package org.apache.hop.pipeline.transforms.maskfields;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -86,6 +87,55 @@ void rejectsPatternsThatDoNotFitTheField() throws Exception {
assertTrue(messages(remarks).stream().anyMatch(text -> text.contains("Missing")));
}
+ @Test
+ void warnsAboutPlainTextKeysAndLongReplacements() throws Exception {
+ MemoryMetadataProvider provider = new MemoryMetadataProvider();
+ MaskingPattern plain = synthetic("Plain", MaskingToken.SEQUENCE);
+ plain.setStorage(MaskingStorage.DATABASE);
+ plain.setConnection("db");
+ plain.setTableName("mask_map");
+ save(provider, plain);
+ MaskingPattern hashed = synthetic("Hashed", MaskingToken.UUID);
+ hashed.setStorage(MaskingStorage.DATABASE);
+ hashed.setConnection("db");
+ hashed.setTableName("mask_map");
+ hashed.setHashSecret("${SECRET}");
+ hashed.setPrefix("${PREFIX}");
+ save(provider, hashed);
+
+ MaskFieldsMeta meta = new MaskFieldsMeta();
+ meta.getFields().add(new MaskField("name", "Plain"));
+ meta.getFields().add(new MaskField("city", "Hashed"));
+ RowMeta prev = new RowMeta();
+ prev.addValueMeta(new ValueMetaString("name"));
+ prev.addValueMeta(new ValueMetaString("city"));
+ Variables variables = new Variables();
+ variables.setVariable("PREFIX", "x".repeat(250));
+
+ List remarks = new ArrayList<>();
+ TransformMeta transform = new TransformMeta();
+ transform.setName("Mask");
+ meta.check(
+ remarks,
+ null,
+ transform,
+ prev,
+ new String[] {"in"},
+ new String[0],
+ null,
+ variables,
+ provider);
+
+ List warnings =
+ remarks.stream()
+ .filter(remark -> remark.getType() == ICheckResult.TYPE_RESULT_WARNING)
+ .map(ICheckResult::getText)
+ .toList();
+ assertEquals(2, warnings.size(), warnings.toString());
+ assertTrue(warnings.get(0).contains("Plain") && warnings.get(0).contains("plain text"));
+ assertTrue(warnings.get(1).contains("Hashed") && warnings.get(1).contains("286"));
+ }
+
@Test
void roundTripsTheTransformAndThePattern() throws Exception {
MaskFieldsMeta meta = new MaskFieldsMeta();
@@ -101,6 +151,9 @@ void roundTripsTheTransformAndThePattern() throws Exception {
pattern.setPrefix("first-name-");
pattern.setPiiClassification("Direct identifier");
pattern.setStorage(MaskingStorage.MEMORY);
+ pattern.setTrimKey(true);
+ pattern.setIgnoreCase(true);
+ pattern.setHashSecret("s3cret");
MemoryMetadataProvider provider = new MemoryMetadataProvider();
JsonMetadataParser parser =
new JsonMetadataParser<>(MaskingPattern.class, provider);
@@ -113,6 +166,10 @@ void roundTripsTheTransformAndThePattern() throws Exception {
assertEquals("Direct identifier", loaded.getPiiClassification());
assertEquals(MaskingStorage.MEMORY, loaded.getStorage());
assertEquals(MaskingValueSource.SYNTHETIC, loaded.getValueSource());
+ assertTrue(loaded.isTrimKey());
+ assertTrue(loaded.isIgnoreCase());
+ assertEquals("s3cret", loaded.getHashSecret());
+ assertFalse(json.toJSONString().contains("s3cret"));
assertNull(MaskingRules.incompatibility(new ValueMetaString("name"), loaded));
}
}
diff --git a/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingEngineTest.java b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingEngineTest.java
index bc4a51ac094..1a670ca33ad 100644
--- a/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingEngineTest.java
+++ b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingEngineTest.java
@@ -17,15 +17,19 @@
package org.apache.hop.pipeline.transforms.maskfields;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
import java.util.UUID;
+import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.row.RowMeta;
import org.apache.hop.core.row.value.ValueMetaString;
import org.apache.hop.pipeline.transforms.maskfields.MaskingEngine.Binding;
+import org.apache.hop.pipeline.transforms.maskfields.store.IMaskingStore;
import org.apache.hop.pipeline.transforms.maskfields.store.MemoryMaskingStore;
import org.junit.jupiter.api.Test;
@@ -94,6 +98,45 @@ void setNullAndSetEmpty() throws Exception {
assertEquals("", masked[1]);
}
+ @Test
+ void aFailingFieldClearsItselfAndTheFieldsAfterIt() throws Exception {
+ MaskingPattern first = synthetic("First name", MaskingStorage.NONE, "fn-", "", "1");
+ MaskingPattern broken = synthetic("Broken", MaskingStorage.MEMORY, "x-", "", "1");
+ IMaskingStore failing =
+ new MemoryMaskingStore() {
+ @Override
+ public synchronized String findOrCreate(
+ String patternName, String sourceKey, MaskAllocator allocator) throws HopException {
+ throw new HopException("mapping table is gone");
+ }
+ };
+ MaskingEngine engine =
+ engine(
+ binding("first", first, null),
+ new Binding("last", broken, "x-", "", 1, failing),
+ binding("city", first, null));
+ RowMeta rowMeta = stringRow("first", "last", "city", "untouched");
+ Object[] row = {"Matt", "Casters", "Gent", "kept"};
+
+ assertThrows(HopException.class, () -> engine.apply(rowMeta, row));
+ assertArrayEquals(new Object[] {"fn-1", null, null, "kept"}, row);
+ }
+
+ @Test
+ void aRememberedValueIgnoresCaseAndSpacesWhenAsked() throws Exception {
+ MaskingPattern pattern = synthetic("First name", MaskingStorage.MEMORY, "fn-", "", "1");
+ MemoryMaskingStore store = new MemoryMaskingStore();
+ long start = 1L;
+ MaskingEngine engine =
+ engine(
+ new Binding(
+ "name", pattern, "fn-", "", start, store, null, new MaskingKey(true, true, null)));
+ RowMeta rowMeta = stringRow("name");
+ assertEquals("fn-1", apply(engine, rowMeta, "Matt")[0]);
+ assertEquals("fn-1", apply(engine, rowMeta, " MATT ")[0]);
+ assertEquals(" ", apply(engine, rowMeta, " ")[0]);
+ }
+
private static MaskingPattern synthetic(
String name, MaskingStorage storage, String prefix, String suffix, String start) {
MaskingPattern pattern = new MaskingPattern();
@@ -107,7 +150,7 @@ private static MaskingPattern synthetic(
return pattern;
}
- private static Binding binding(String field, MaskingPattern pattern, MemoryMaskingStore store) {
+ private static Binding binding(String field, MaskingPattern pattern, IMaskingStore store) {
long start = 1L;
if (pattern.getSequenceStart() != null && !pattern.getSequenceStart().isBlank()) {
start = Long.parseLong(pattern.getSequenceStart().trim());
diff --git a/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingKeyTest.java b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingKeyTest.java
new file mode 100644
index 00000000000..d57ddea2296
--- /dev/null
+++ b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingKeyTest.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.maskfields;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.math.BigDecimal;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.value.ValueMetaBigNumber;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaNumber;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.variables.Variables;
+import org.junit.jupiter.api.Test;
+
+class MaskingKeyTest {
+
+ @Test
+ void aNumberKeyDoesNotDependOnTheConversionMask() throws Exception {
+ MaskingKey key = MaskingKey.plain();
+ ValueMetaNumber plain = new ValueMetaNumber("amount");
+ ValueMetaNumber formatted = new ValueMetaNumber("amount");
+ formatted.setConversionMask("#,##0.00");
+ assertEquals("1234.5", key.normalize(plain, 1234.5));
+ assertEquals("1234.5", key.normalize(formatted, 1234.5));
+
+ ValueMetaInteger integer = new ValueMetaInteger("id");
+ integer.setConversionMask("000000");
+ assertEquals("42", key.normalize(integer, 42L));
+ assertEquals("000042", MaskingKey.legacyKey(integer, 42L));
+
+ assertEquals("10", key.normalize(new ValueMetaBigNumber("big"), new BigDecimal("10.000")));
+ }
+
+ @Test
+ void trimAndIgnoreCaseAreOptional() throws Exception {
+ ValueMetaString name = new ValueMetaString("name");
+ assertEquals(" Matt ", MaskingKey.plain().normalize(name, " Matt "));
+ assertEquals("matt", new MaskingKey(true, true, null).normalize(name, " Matt "));
+ assertEquals("", new MaskingKey(true, false, null).normalize(name, " "));
+ }
+
+ @Test
+ void aHashSecretThatDoesNotResolveIsAnError() {
+ Variables variables = new Variables();
+ variables.setVariable("EMPTY_SECRET", "");
+ variables.setVariable("BROKEN_SECRET", "Encrypted not-a-ciphertext");
+ for (String secret :
+ new String[] {
+ "${NOT_SET}", "${EMPTY_SECRET}", "%%NOT_SET%%", "${BROKEN_SECRET}", "Encrypted "
+ }) {
+ MaskingPattern pattern = databasePattern(secret);
+ HopException e =
+ assertThrows(HopException.class, () -> MaskingKey.forPattern(pattern, variables));
+ assertTrue(e.getMessage().contains("Names"), e.getMessage());
+ }
+ }
+
+ @Test
+ void aResolvedHashSecretHashesDatabaseKeysOnly() throws Exception {
+ Variables variables = new Variables();
+ variables.setVariable("SECRET", "s3cret");
+ String expected = new MaskingKey(false, false, "s3cret").storeKey("Matt");
+ assertEquals(
+ expected, MaskingKey.forPattern(databasePattern("${SECRET}"), variables).storeKey("Matt"));
+
+ MaskingPattern memory = databasePattern("${NOT_SET}");
+ memory.setStorage(MaskingStorage.MEMORY);
+ assertEquals("Matt", MaskingKey.forPattern(memory, variables).storeKey("Matt"));
+ assertEquals("Matt", MaskingKey.forPattern(databasePattern(""), variables).storeKey("Matt"));
+ }
+
+ private static MaskingPattern databasePattern(String secret) {
+ MaskingPattern pattern = new MaskingPattern();
+ pattern.setName("Names");
+ pattern.setStorage(MaskingStorage.DATABASE);
+ pattern.setHashSecret(secret);
+ return pattern;
+ }
+
+ @Test
+ void aSecretHashesTheStoredKey() throws Exception {
+ MaskingKey key = new MaskingKey(false, false, "s3cret");
+ String hashed = key.storeKey("Matt");
+ assertTrue(hashed.startsWith(MaskingKey.HMAC_PREFIX));
+ assertEquals(MaskingKey.HMAC_PREFIX.length() + 64, hashed.length());
+ assertEquals(hashed, new MaskingKey(false, false, "s3cret").storeKey("Matt"));
+ assertNotEquals(hashed, new MaskingKey(false, false, "other").storeKey("Matt"));
+ assertEquals("Matt", MaskingKey.plain().storeKey("Matt"));
+ }
+}
diff --git a/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingRuntimeTest.java b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingRuntimeTest.java
index e5627d77849..deb5a1fe363 100644
--- a/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingRuntimeTest.java
+++ b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/MaskingRuntimeTest.java
@@ -18,16 +18,27 @@
package org.apache.hop.pipeline.transforms.maskfields;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
import java.util.List;
import java.util.UUID;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.database.DatabaseMeta;
import org.apache.hop.core.row.RowMeta;
import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.variables.Variables;
import org.apache.hop.pipeline.transforms.maskfields.MaskingEngine.Binding;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
class MaskingRuntimeTest {
+ @BeforeAll
+ static void initHop() throws Exception {
+ HopEnvironment.init();
+ }
+
@Test
void memoryIsSharedInsideOneExecutionAndResetForTheNext() throws Exception {
String executionId = uniqueId();
@@ -125,6 +136,23 @@ private static RowMeta stringRow(String name) {
return rowMeta;
}
+ @Test
+ void aManualUrlIsKeyedOnTheUrlAConnectionOpens() throws Exception {
+ DatabaseMeta databaseMeta =
+ new DatabaseMeta("mask", "H2", "Native", "", "unused", "", "sa", "");
+ databaseMeta.setManualUrl("jdbc:h2:mem:${MASK_DB}");
+ Variables one = new Variables();
+ one.setVariable("MASK_DB", "one");
+ Variables two = new Variables();
+ two.setVariable("MASK_DB", "two");
+
+ String key = MaskingRuntime.databaseKey(one, databaseMeta, null, "mask_map");
+ assertEquals("jdbc:h2:mem:one", key.substring(0, key.indexOf('\0')));
+ assertNotEquals(key, MaskingRuntime.databaseKey(two, databaseMeta, null, "mask_map"));
+ // Unresolved, the URL does not say which database it opens.
+ assertNull(MaskingRuntime.databaseKey(new Variables(), databaseMeta, null, "mask_map"));
+ }
+
private static String uniqueId() {
return "runtime-" + UUID.randomUUID();
}
diff --git a/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/store/DatabaseMaskingStoreTest.java b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/store/DatabaseMaskingStoreTest.java
index 5695dd17e1f..84ae5abcbed 100644
--- a/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/store/DatabaseMaskingStoreTest.java
+++ b/plugins/transforms/maskfields/src/test/java/org/apache/hop/pipeline/transforms/maskfields/store/DatabaseMaskingStoreTest.java
@@ -18,13 +18,29 @@
package org.apache.hop.pipeline.transforms.maskfields.store;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
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 java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.database.Database;
import org.apache.hop.core.database.DatabaseMeta;
+import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.logging.ILoggingObject;
import org.apache.hop.core.variables.Variables;
+import org.apache.hop.pipeline.transforms.maskfields.MaskingKey;
import org.apache.hop.pipeline.transforms.maskfields.MaskingRuntime;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -124,8 +140,221 @@ void runtimeSharesOneDatabaseStoreAcrossCopies() throws Exception {
}
}
+ @Test
+ void twoProcessesNeverHandOutTheSameValue() throws Exception {
+ DatabaseMeta databaseMeta = h2();
+ Variables variables = new Variables();
+ // Two stores on one table stand in for two Hop processes: no shared lock, two connections.
+ DatabaseMaskingStore first =
+ new DatabaseMaskingStore((ILoggingObject) null, variables, databaseMeta, null, "mask_map");
+ DatabaseMaskingStore second =
+ new DatabaseMaskingStore((ILoggingObject) null, variables, databaseMeta, null, "mask_map");
+ first.open();
+ second.open();
+ int perStore = 200;
+ // The first value creates the sequence row. Both processes then race on updating it.
+ Set masked = new HashSet<>(List.of(allocate(first, "seed")));
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ Future> a = executor.submit(() -> allocateAll(first, "a", perStore));
+ Future> b = executor.submit(() -> allocateAll(second, "b", perStore));
+ masked.addAll(a.get(60, TimeUnit.SECONDS));
+ masked.addAll(b.get(60, TimeUnit.SECONDS));
+ assertEquals(2 * perStore + 1, masked.size());
+ } finally {
+ executor.shutdownNow();
+ first.close();
+ second.close();
+ }
+ }
+
+ @Test
+ void aHashSecretKeepsOriginalValuesOutOfTheTable() throws Exception {
+ DatabaseMeta databaseMeta = h2();
+ Variables variables = new Variables();
+ MaskingKey key = new MaskingKey(false, false, "s3cret");
+ DatabaseMaskingStore store =
+ new DatabaseMaskingStore((ILoggingObject) null, variables, databaseMeta, null, "mask_map");
+ store.open();
+ try {
+ String hashed = key.storeKey("Matt");
+ assertTrue(hashed.startsWith("hmac-sha256:"));
+ assertEquals("first-name-1", allocate(store, hashed, "Matt"));
+ assertEquals("first-name-1", allocate(store, hashed, "Matt"));
+ assertEquals(List.of(hashed), sourceKeys(databaseMeta, variables));
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ void aPlainTextRowMovesToItsHashedKey() throws Exception {
+ DatabaseMeta databaseMeta = h2();
+ Variables variables = new Variables();
+ DatabaseMaskingStore store =
+ new DatabaseMaskingStore((ILoggingObject) null, variables, databaseMeta, null, "mask_map");
+ store.open();
+ try {
+ // A row written before the pattern had a hash secret.
+ assertEquals("first-name-1", allocate(store, "Matt"));
+ } finally {
+ store.close();
+ }
+
+ String hashed = new MaskingKey(false, false, "s3cret").storeKey("Matt");
+ DatabaseMaskingStore later =
+ new DatabaseMaskingStore((ILoggingObject) null, variables, databaseMeta, null, "mask_map");
+ later.open();
+ try {
+ assertEquals("first-name-1", allocate(later, hashed, "Matt"));
+ assertEquals("first-name-2", allocate(later, hashed + "x", "Ann"));
+ } finally {
+ later.close();
+ }
+ List keys = sourceKeys(databaseMeta, variables);
+ assertTrue(keys.contains(hashed));
+ assertFalse(keys.contains("Matt"));
+ }
+
+ @Test
+ void twoStoredSpellingsOfOneKeyLeaveNoPlainTextRow() throws Exception {
+ DatabaseMeta databaseMeta = h2();
+ Variables variables = new Variables();
+ DatabaseMaskingStore store =
+ new DatabaseMaskingStore((ILoggingObject) null, variables, databaseMeta, null, "mask_map");
+ store.open();
+ try {
+ // Rows written before the pattern ignored case.
+ assertEquals("first-name-1", allocate(store, "Matt"));
+ assertEquals("first-name-2", allocate(store, "MATT"));
+ } finally {
+ store.close();
+ }
+
+ DatabaseMaskingStore later =
+ new DatabaseMaskingStore((ILoggingObject) null, variables, databaseMeta, null, "mask_map");
+ later.open();
+ try {
+ assertEquals("first-name-1", allocate(later, "matt", "Matt"));
+ // The new key is cached now. The second spelling still has to leave the table.
+ assertEquals("first-name-1", allocate(later, "matt", "MATT"));
+ } finally {
+ later.close();
+ }
+ assertEquals(List.of("matt"), sourceKeys(databaseMeta, variables));
+
+ DatabaseMaskingStore again =
+ new DatabaseMaskingStore((ILoggingObject) null, variables, databaseMeta, null, "mask_map");
+ again.open();
+ try {
+ assertEquals("first-name-1", allocate(again, "matt", "MATT"));
+ } finally {
+ again.close();
+ }
+ assertEquals(List.of("matt"), sourceKeys(databaseMeta, variables));
+ }
+
+ @Test
+ void aManualUrlIsKeyedOnTheDatabaseItOpens() throws Exception {
+ String one = "mem:mask" + UUID.randomUUID().toString().replace("-", "");
+ String two = "mem:mask" + UUID.randomUUID().toString().replace("-", "");
+ DatabaseMeta databaseMeta = h2();
+ databaseMeta.setManualUrl("jdbc:h2:${MASK_DB};DB_CLOSE_DELAY=-1");
+ Variables first = new Variables();
+ first.setVariable("MASK_DB", one);
+ Variables second = new Variables();
+ second.setVariable("MASK_DB", two);
+
+ MaskingRuntime.Lease lease = MaskingRuntime.getInstance().acquire("db-" + UUID.randomUUID());
+ try {
+ DatabaseMaskingStore firstStore = lease.database(null, first, databaseMeta, null, "mask_map");
+ DatabaseMaskingStore secondStore =
+ lease.database(null, second, databaseMeta, null, "mask_map");
+ assertNotSame(firstStore, secondStore);
+ assertEquals("first-name-1", allocate(firstStore, "Matt"));
+ assertEquals("first-name-1", allocate(secondStore, "Ann"));
+ } finally {
+ lease.release();
+ }
+ }
+
+ @Test
+ void connectionsWithTheSameNameOnDifferentDatabasesGetTheirOwnStore() throws Exception {
+ Variables variables = new Variables();
+ DatabaseMeta one = h2();
+ DatabaseMeta two = h2();
+ two.setName(one.getName());
+ MaskingRuntime.Lease lease = MaskingRuntime.getInstance().acquire("db-" + UUID.randomUUID());
+ try {
+ DatabaseMaskingStore first = lease.database(null, variables, one, null, "mask_map");
+ DatabaseMaskingStore second = lease.database(null, variables, two, null, "mask_map");
+ assertNotSame(first, second);
+ assertEquals("first-name-1", allocate(first, "Matt"));
+ assertEquals("first-name-1", allocate(second, "Ann"));
+ } finally {
+ lease.release();
+ }
+ }
+
+ @Test
+ void anExhaustedRetryKeepsTheSourceKeyOutOfTheError() throws Exception {
+ DatabaseMaskingStore store =
+ new DatabaseMaskingStore((ILoggingObject) null, new Variables(), h2(), null, "mask_map");
+ store.open();
+ try {
+ // Every new key gets the same replacement, so the unique key on it rejects every attempt.
+ assertEquals("same", store.findOrCreate("First name", "Matt", current -> "same"));
+ HopException e =
+ assertThrows(
+ HopException.class, () -> store.findOrCreate("First name", "Ann", current -> "same"));
+ assertNull(e.getCause());
+ assertFalse(e.getMessage().contains("Ann"), e.getMessage());
+ assertTrue(e.getMessage().contains("First name"), e.getMessage());
+ } finally {
+ store.close();
+ }
+ }
+
+ private static List sourceKeys(DatabaseMeta databaseMeta, Variables variables)
+ throws Exception {
+ Database database = new Database((ILoggingObject) null, variables, databaseMeta);
+ database.connect();
+ try {
+ List keys = new ArrayList<>();
+ for (Object[] row : database.getRows("SELECT source_key FROM mask_map", 1000)) {
+ keys.add((String) row[0]);
+ }
+ return keys;
+ } finally {
+ database.disconnect();
+ }
+ }
+
+ private static List allocateAll(DatabaseMaskingStore store, String prefix, int count)
+ throws Exception {
+ List masked = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ masked.add(allocate(store, prefix + i));
+ }
+ return masked;
+ }
+
+ private static DatabaseMeta h2() {
+ String id = UUID.randomUUID().toString().replace("-", "");
+ return new DatabaseMeta(
+ "mask-" + id, "H2", "Native", "", "mem:mask" + id + ";DB_CLOSE_DELAY=-1", "", "sa", "");
+ }
+
private static String allocate(DatabaseMaskingStore store, String source) throws Exception {
+ return allocate(store, source, null);
+ }
+
+ private static String allocate(DatabaseMaskingStore store, String source, String legacy)
+ throws Exception {
return store.findOrCreate(
- "First name", source, current -> "first-name-" + current.allocateSequence("First name", 1));
+ "First name",
+ source,
+ legacy == null ? null : () -> legacy,
+ current -> "first-name-" + current.allocateSequence("First name", 1));
}
}