From e926a179689a777f81823225b2366415b48e0475 Mon Sep 17 00:00:00 2001 From: Mark Payne Date: Tue, 28 Jul 2020 13:44:32 -0400 Subject: [PATCH] NIFI-7683: Fixed bug that resulted in a byte[] being allocated and held onto by a member variable that was unprotected. This caused the byte array to be modified by two different threads concurrently, which can potentially cause corruption of FlowFile's data Signed-off-by: Matthew Burgess This closes #4439 --- .../nifi/processors/standard/ReplaceText.java | 73 ++++++------------- 1 file changed, 22 insertions(+), 51 deletions(-) diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java index 22ba5f8c90..97d9cfc8f0 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java @@ -46,8 +46,6 @@ import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.ProcessorInitializationContext; import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.exception.ProcessException; -import org.apache.nifi.processor.io.InputStreamCallback; -import org.apache.nifi.processor.io.OutputStreamCallback; import org.apache.nifi.processor.io.StreamCallback; import org.apache.nifi.processor.util.StandardValidators; import org.apache.nifi.stream.io.StreamUtils; @@ -260,14 +258,6 @@ public class ReplaceText extends AbstractProcessor { public void setup(ProcessContext context) { final String replacementStrategy = context.getProperty(REPLACEMENT_STRATEGY).getValue(); final String evaluateMode = context.getProperty(EVALUATION_MODE).getValue(); - final int maxBufferSize = context.getProperty(MAX_BUFFER_SIZE).asDataSize(DataUnit.B).intValue(); - - final byte[] buffer; - if (replacementStrategy.equalsIgnoreCase(regexReplaceValue) || replacementStrategy.equalsIgnoreCase(literalReplaceValue)) { - buffer = new byte[maxBufferSize]; - } else { - buffer = null; - } switch (replacementStrategy) { case prependValue: @@ -285,12 +275,13 @@ public class ReplaceText extends AbstractProcessor { && context.getProperty(REPLACEMENT_VALUE).getValue().isEmpty()) { replacementStrategyExecutor = new AlwaysReplace(); } else { - replacementStrategyExecutor = new RegexReplace(buffer, context); + final String regex = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions().getValue(); + replacementStrategyExecutor = new RegexReplace(regex); } break; case literalReplaceValue: - replacementStrategyExecutor = new LiteralReplace(buffer); + replacementStrategyExecutor = new LiteralReplace(); break; case alwaysReplace: replacementStrategyExecutor = new AlwaysReplace(); @@ -382,7 +373,7 @@ public class ReplaceText extends AbstractProcessor { return value; } - private class AlwaysReplace implements ReplacementStrategyExecutor { + private static class AlwaysReplace implements ReplacementStrategyExecutor { @Override public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { @@ -390,12 +381,7 @@ public class ReplaceText extends AbstractProcessor { final StringBuilder lineEndingBuilder = new StringBuilder(2); if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { - flowFile = session.write(flowFile, new StreamCallback() { - @Override - public void process(final InputStream in, final OutputStream out) throws IOException { - out.write(replacementValue.getBytes(charset)); - } - }); + flowFile = session.write(flowFile, out -> out.write(replacementValue.getBytes(charset))); } else { flowFile = session.write(flowFile, new StreamReplaceCallback(charset, maxBufferSize, context.getProperty(LINE_BY_LINE_EVALUATION_MODE).getValue(), ((bw, oneLine) -> { @@ -427,7 +413,7 @@ public class ReplaceText extends AbstractProcessor { } } - private class PrependReplace implements ReplacementStrategyExecutor { + private static class PrependReplace implements ReplacementStrategyExecutor { @Override public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue(); @@ -454,7 +440,7 @@ public class ReplaceText extends AbstractProcessor { } - private class AppendReplace implements ReplacementStrategyExecutor { + private static class AppendReplace implements ReplacementStrategyExecutor { @Override public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { @@ -506,8 +492,7 @@ public class ReplaceText extends AbstractProcessor { } - private class RegexReplace implements ReplacementStrategyExecutor { - private final byte[] buffer; + private static class RegexReplace implements ReplacementStrategyExecutor { private final int numCapturingGroups; private final Map additionalAttrs; @@ -520,11 +505,8 @@ public class ReplaceText extends AbstractProcessor { } }; - public RegexReplace(final byte[] buffer, final ProcessContext context) { - this.buffer = buffer; - - final String regexValue = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions().getValue(); - numCapturingGroups = Pattern.compile(regexValue).matcher("").groupCount(); + public RegexReplace(final String regex) { + numCapturingGroups = Pattern.compile(regex).matcher("").groupCount(); additionalAttrs = new HashMap<>(numCapturingGroups); } @@ -535,15 +517,13 @@ public class ReplaceText extends AbstractProcessor { final String searchRegex = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue(); final Pattern searchPattern = Pattern.compile(searchRegex); - final int flowFileSize = (int) flowFile.getSize(); FlowFile updatedFlowFile; if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { - session.read(flowFile, new InputStreamCallback() { - @Override - public void process(final InputStream in) throws IOException { - StreamUtils.fillBuffer(in, buffer, false); - } - }); + final int flowFileSize = (int) flowFile.getSize(); + final int bufferSize = Math.min(maxBufferSize, flowFileSize); + final byte[] buffer = new byte[bufferSize]; + + session.read(flowFile, in -> StreamUtils.fillBuffer(in, buffer, false)); final String contentString = new String(buffer, 0, flowFileSize, charset); additionalAttrs.clear(); @@ -571,12 +551,7 @@ public class ReplaceText extends AbstractProcessor { matcher.appendTail(sb); final String updatedValue = sb.toString(); - updatedFlowFile = session.write(flowFile, new OutputStreamCallback() { - @Override - public void process(final OutputStream out) throws IOException { - out.write(updatedValue.getBytes(charset)); - } - }); + updatedFlowFile = session.write(flowFile, out -> out.write(updatedValue.getBytes(charset))); } else { return flowFile; } @@ -626,22 +601,18 @@ public class ReplaceText extends AbstractProcessor { } } - private class LiteralReplace implements ReplacementStrategyExecutor { - private final byte[] buffer; - - public LiteralReplace(final byte[] buffer) { - this.buffer = buffer; - } - + private static class LiteralReplace implements ReplacementStrategyExecutor { @Override public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { - final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue(); final AttributeValueDecorator quotedAttributeDecorator = Pattern::quote; final String searchValue = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue(); - final int flowFileSize = (int) flowFile.getSize(); if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { + final int flowFileSize = (int) flowFile.getSize(); + final int bufferSize = Math.min(maxBufferSize, flowFileSize); + final byte[] buffer = new byte[bufferSize]; + flowFile = session.write(flowFile, new StreamCallback() { @Override public void process(final InputStream in, final OutputStream out) throws IOException { @@ -710,7 +681,7 @@ public class ReplaceText extends AbstractProcessor { } - private class StreamReplaceCallback implements StreamCallback { + private static class StreamReplaceCallback implements StreamCallback { private final Charset charset; private final int maxBufferSize; private final String lineByLineEvaluationMode;