This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 3f4162f2aec NIFI-15718 Validate ZIP CRC-32 checksums in UnpackContent
(#11636)
3f4162f2aec is described below
commit 3f4162f2aec077415c4490efc3b790dd5d37c383
Author: Joe Witt <[email protected]>
AuthorDate: Wed Sep 9 09:37:45 2026 -0700
NIFI-15718 Validate ZIP CRC-32 checksums in UnpackContent (#11636)
Signed-off-by: David Handermann <[email protected]>
---
.../nifi/processors/standard/UnpackContent.java | 166 ++++++++---
.../processors/standard/TestUnpackContent.java | 311 ++++++++++++++++++++-
2 files changed, 434 insertions(+), 43 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java
index 5aa097ae6fb..c0b0c777570 100644
---
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java
@@ -78,17 +78,23 @@ import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;
+import java.util.zip.CRC32;
+import java.util.zip.CheckedOutputStream;
@SideEffectFree
@SupportsBatching
@InputRequirement(Requirement.INPUT_REQUIRED)
@Tags({"Unpack", "un-merge", "tar", "zip", "archive", "flowfile-stream",
"flowfile-stream-v3"})
-@CapabilityDescription("Unpacks the content of FlowFiles that have been
packaged with one of several different Packaging Formats, emitting one to many "
- + "FlowFiles for each input FlowFile. Supported formats are TAR, ZIP,
and FlowFile Stream packages.")
+@CapabilityDescription("""
+ Unpacks the content of FlowFiles that have been packaged with one of
several different Packaging Formats, emitting one to many \
+ FlowFiles for each input FlowFile. Supported formats are TAR, ZIP, and
FlowFile Stream packages. For ZIP, every entry that provides a CRC-32 \
+ checksum is validated against the uncompressed bytes, including
entries that are not extracted because of File Filter. A mismatch means the \
+ archive is corrupt: any FlowFiles already created for earlier entries
are removed and the original FlowFile is routed to failure.""")
@ReadsAttribute(attribute = "mime.type", description = "If the <Packaging
Format> property is set to use mime.type attribute, this attribute is used "
+ "to determine the FlowFile's MIME Type. In this case, if the
attribute is set to application/tar, the TAR Packaging Format will be used. If "
+ "the attribute is set to application/zip, the ZIP Packaging Format
will be used. If the attribute is set to application/flowfile-v3 or "
@@ -174,7 +180,9 @@ public class UnpackContent extends AbstractProcessor {
public static final PropertyDescriptor FILE_FILTER = new
PropertyDescriptor.Builder()
.name("File Filter")
- .description("Only files contained in the archive whose names
match the given regular expression will be extracted (tar/zip only)")
+ .description("""
+ Only files contained in the archive whose names match the
given regular expression will be extracted (tar/zip only). \
+ ZIP CRC-32 checksums are still validated for every entry
that provides one, even when the entry is not extracted.""")
.required(true)
.defaultValue(".*")
.addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
@@ -182,7 +190,10 @@ public class UnpackContent extends AbstractProcessor {
public static final PropertyDescriptor PASSWORD = new
PropertyDescriptor.Builder()
.name("Password")
- .description("Password used for decrypting Zip archives encrypted
with ZipCrypto or AES. Configuring a password disables support for alternative
Zip compression algorithms.")
+ .description("""
+ Password used for decrypting Zip archives encrypted with
ZipCrypto or AES. Configuring a password disables support for alternative \
+ Zip compression algorithms. CRC-32 validation is not
applied when a password is configured because the encrypted Zip reader does \
+ not expose the uncompressed checksum.""")
.required(false)
.sensitive(true)
.dependsOn(PACKAGING_FORMAT, PackageFormat.ZIP_FORMAT,
PackageFormat.AUTO_DETECT_FORMAT)
@@ -220,7 +231,7 @@ public class UnpackContent extends AbstractProcessor {
.build();
public static final Relationship REL_FAILURE = new Relationship.Builder()
.name("failure")
- .description("The original FlowFile is sent to this relationship
when it cannot be unpacked for some reason")
+ .description("The original FlowFile is sent to this relationship
when it cannot be unpacked for some reason, including a ZIP CRC-32 mismatch")
.build();
private static final Set<Relationship> RELATIONSHIPS = Set.of(
@@ -332,7 +343,14 @@ public class UnpackContent extends AbstractProcessor {
final List<FlowFile> unpacked = new ArrayList<>();
try {
- unpacker.unpack(session, flowFile, unpacked);
+ final UnpackResult result = unpacker.unpack(session, flowFile,
unpacked);
+ if (result.failed()) {
+ logger.error("Unable to unpack {} because {}; routing to
failure", flowFile, result.failureReason());
+ session.transfer(flowFile, REL_FAILURE);
+ session.remove(unpacked);
+ return;
+ }
+
if (unpacked.isEmpty()) {
logger.error("Unable to unpack {} because it does not appear
to have any entries; routing to failure", flowFile);
session.transfer(flowFile, REL_FAILURE);
@@ -377,7 +395,12 @@ public class UnpackContent extends AbstractProcessor {
this.fileFilter = fileFilter;
}
- abstract void unpack(ProcessSession session, FlowFile source,
List<FlowFile> unpacked);
+ /**
+ * Unpacks the source FlowFile, adding extracted children to {@code
unpacked}.
+ *
+ * @return success, or a validation failure such as a ZIP CRC mismatch
+ */
+ abstract UnpackResult unpack(ProcessSession session, FlowFile source,
List<FlowFile> unpacked);
protected boolean fileMatches(final ArchiveEntry entry) {
return fileMatches(entry.getName());
@@ -388,13 +411,32 @@ public class UnpackContent extends AbstractProcessor {
}
}
+ /**
+ * Outcome of unpacking one archive. {@link #failed()} is an expected
validation failure, not an unexpected error.
+ */
+ private record UnpackResult(String failureReason) {
+ private static final UnpackResult SUCCESS = new UnpackResult(null);
+
+ private static UnpackResult success() {
+ return SUCCESS;
+ }
+
+ private static UnpackResult failure(final String reason) {
+ return new UnpackResult(reason);
+ }
+
+ private boolean failed() {
+ return failureReason != null;
+ }
+ }
+
private static class TarUnpacker extends Unpacker {
public TarUnpacker(Pattern fileFilter) {
super(fileFilter);
}
@Override
- public void unpack(final ProcessSession session, final FlowFile
source, final List<FlowFile> unpacked) {
+ public UnpackResult unpack(final ProcessSession session, final
FlowFile source, final List<FlowFile> unpacked) {
final String fragmentId = UUID.randomUUID().toString();
final Map<String, String> attributes = new HashMap<>();
session.read(source, inputStream -> {
@@ -452,6 +494,7 @@ public class UnpackContent extends AbstractProcessor {
}
}
});
+ return UnpackResult.success();
}
}
@@ -467,13 +510,19 @@ public class UnpackContent extends AbstractProcessor {
}
@Override
- public void unpack(final ProcessSession session, final FlowFile
source, final List<FlowFile> unpacked) {
+ public UnpackResult unpack(final ProcessSession session, final
FlowFile source, final List<FlowFile> unpacked) {
final String fragmentId = UUID.randomUUID().toString();
if (password == null) {
- session.read(source, new
CompressedZipInputStreamCallback(fileFilter, session, source, unpacked,
fragmentId, allowStoredEntriesWithDataDescriptor, filenameEncoding));
- } else {
- session.read(source, new
EncryptedZipInputStreamCallback(fileFilter, session, source, unpacked,
fragmentId, password, filenameEncoding));
+ final CompressedZipInputStreamCallback callback = new
CompressedZipInputStreamCallback(fileFilter, session, source, unpacked,
fragmentId,
+ allowStoredEntriesWithDataDescriptor,
filenameEncoding);
+ session.read(source, callback);
+ return callback.getCrcMismatch()
+ .map(UnpackResult::failure)
+ .orElseGet(UnpackResult::success);
}
+
+ session.read(source, new
EncryptedZipInputStreamCallback(fileFilter, session, source, unpacked,
fragmentId, password, filenameEncoding));
+ return UnpackResult.success();
}
private abstract static class ZipInputStreamCallback implements
InputStreamCallback {
@@ -509,24 +558,36 @@ public class UnpackContent extends AbstractProcessor {
return !directory && (fileFilter == null ||
fileFilter.matcher(fileName).find());
}
- protected void processEntry(final InputStream zipInputStream,
boolean directory, String zipEntryName, Map<String, String> attributes) {
- if (isFileEntryMatched(directory, zipEntryName)) {
- final File file = new File(zipEntryName);
- final String parentDirectory = (file.getParent() == null)
? PATH_SEPARATOR : file.getParent();
-
- FlowFile unpackedFile = session.create(sourceFlowFile);
- try {
- attributes.put(CoreAttributes.FILENAME.key(),
file.getName());
- attributes.put(CoreAttributes.PATH.key(),
parentDirectory);
- attributes.put(CoreAttributes.MIME_TYPE.key(),
OCTET_STREAM);
- attributes.put(FRAGMENT_ID, fragmentId);
- attributes.put(FRAGMENT_INDEX,
String.valueOf(++fragmentIndex));
- unpackedFile = session.putAllAttributes(unpackedFile,
attributes);
- unpackedFile = session.write(unpackedFile,
zipInputStream::transferTo);
- } finally {
- unpacked.add(unpackedFile);
- }
+ /**
+ * Reads the current ZIP entry, updating CRC-32 of the
uncompressed bytes. When the entry is a
+ * matching file, those bytes are also written to a child
FlowFile. Filtered and directory entries
+ * are consumed without creating a FlowFile so their checksum can
still be verified.
+ *
+ * @return CRC-32 of the uncompressed entry bytes
+ */
+ protected long processEntry(final InputStream zipInputStream,
final boolean directory, final String zipEntryName, final Map<String, String>
attributes) throws IOException {
+ final CRC32 crc = new CRC32();
+ if (!isFileEntryMatched(directory, zipEntryName)) {
+ zipInputStream.transferTo(new
CheckedOutputStream(OutputStream.nullOutputStream(), crc));
+ return crc.getValue();
+ }
+
+ final File file = new File(zipEntryName);
+ final String parentDirectory = (file.getParent() == null) ?
PATH_SEPARATOR : file.getParent();
+
+ FlowFile unpackedFile = session.create(sourceFlowFile);
+ try {
+ attributes.put(CoreAttributes.FILENAME.key(),
file.getName());
+ attributes.put(CoreAttributes.PATH.key(), parentDirectory);
+ attributes.put(CoreAttributes.MIME_TYPE.key(),
OCTET_STREAM);
+ attributes.put(FRAGMENT_ID, fragmentId);
+ attributes.put(FRAGMENT_INDEX,
String.valueOf(++fragmentIndex));
+ unpackedFile = session.putAllAttributes(unpackedFile,
attributes);
+ unpackedFile = session.write(unpackedFile, outputStream ->
zipInputStream.transferTo(new CheckedOutputStream(outputStream, crc)));
+ } finally {
+ unpacked.add(unpackedFile);
}
+ return crc.getValue();
}
protected void addFileSizeAttribute(long fileSize, Map<String,
String> attributes) {
@@ -564,11 +625,29 @@ public class UnpackContent extends AbstractProcessor {
}
}
+ /**
+ * Zip entry whose uncompressed CRC has been calculated but whose
stored CRC is not comparable
+ * until Commons Compress closes the entry, which happens on the
following call to getNextEntry().
+ */
+ private record UnverifiedEntry(ZipArchiveEntry entry, long actualCrc) {
+
+ private boolean crcMatches() {
+ final long expectedCrc = entry.getCrc();
+ return expectedCrc == ZipArchiveEntry.CRC_UNKNOWN ||
expectedCrc == actualCrc;
+ }
+
+ private String describeMismatch() {
+ return "CRC mismatch for zip entry '%s': expected 0x%x but
calculated 0x%x".formatted(entry.getName(), entry.getCrc(), actualCrc);
+ }
+ }
+
private static class CompressedZipInputStreamCallback extends
ZipInputStreamCallback {
private final boolean allowStoredEntriesWithDataDescriptor;
private final Charset filenameEncoding;
+ private Optional<String> crcMismatch = Optional.empty();
+
private CompressedZipInputStreamCallback(
final Pattern fileFilter,
final ProcessSession session,
@@ -583,13 +662,32 @@ public class UnpackContent extends AbstractProcessor {
this.filenameEncoding = filenameEncoding;
}
+ private Optional<String> getCrcMismatch() {
+ return crcMismatch;
+ }
+
+ private boolean recordIfCrcMismatch(final UnverifiedEntry
unverifiedEntry) {
+ if (unverifiedEntry != null && !unverifiedEntry.crcMatches()) {
+ crcMismatch =
Optional.of(unverifiedEntry.describeMismatch());
+ return true;
+ }
+ return false;
+ }
+
@Override
public void process(final InputStream inputStream) throws
IOException {
try (final ZipArchiveInputStream zipInputStream = new
ZipArchiveInputStream(new BufferedInputStream(inputStream),
filenameEncoding.toString(), true,
allowStoredEntriesWithDataDescriptor)) {
- ZipArchiveEntry zipEntry;
final Map<String, String> attributes = new HashMap<>();
+ UnverifiedEntry unverifiedEntry = null;
+
+ ZipArchiveEntry zipEntry;
while ((zipEntry = zipInputStream.getNextEntry()) != null)
{
+ // getNextEntry() closed the preceding entry, making
its stored CRC available for comparison
+ if (recordIfCrcMismatch(unverifiedEntry)) {
+ return;
+ }
+
addEncryptionMethodAttribute(EncryptionMethod.NONE,
attributes);
addFileSizeAttribute(zipEntry.getSize(), attributes);
addFilePermissionsAttribute(zipEntry.getUnixMode(),
attributes);
@@ -599,9 +697,12 @@ public class UnpackContent extends AbstractProcessor {
Instant creation = zipEntry.getTime() > 0 ? new
Date(zipEntry.getTime()).toInstant() : null;
Instant lastAccess = zipEntry.getLastAccessTime() !=
null ? zipEntry.getLastAccessTime().toInstant() : null;
addZipEntryTimeAttributes(lastModified, creation,
lastAccess, attributes);
- processEntry(zipInputStream, zipEntry.isDirectory(),
zipEntry.getName(), attributes);
+
+ unverifiedEntry = new UnverifiedEntry(zipEntry,
processEntry(zipInputStream, zipEntry.isDirectory(), zipEntry.getName(),
attributes));
attributes.clear();
}
+
+ recordIfCrcMismatch(unverifiedEntry);
}
}
}
@@ -658,7 +759,7 @@ public class UnpackContent extends AbstractProcessor {
}
@Override
- public void unpack(final ProcessSession session, final FlowFile
source, final List<FlowFile> unpacked) {
+ public UnpackResult unpack(final ProcessSession session, final
FlowFile source, final List<FlowFile> unpacked) {
session.read(source, inputStream -> {
try (final InputStream in = new
BufferedInputStream(inputStream)) {
while (unpackager.hasMoreData()) {
@@ -700,6 +801,7 @@ public class UnpackContent extends AbstractProcessor {
}
}
});
+ return UnpackResult.success();
}
}
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java
index 80727a826b9..cbed7167bbd 100644
---
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java
@@ -23,6 +23,7 @@ import org.apache.commons.io.Charsets;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Strings;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.util.LogMessage;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.PropertyMigrationResult;
import org.apache.nifi.util.TestRunner;
@@ -31,21 +32,28 @@ import org.junit.jupiter.api.Test;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.zip.CRC32;
+import java.util.zip.ZipEntry;
import static org.apache.nifi.processors.standard.SplitContent.FRAGMENT_COUNT;
import static org.apache.nifi.processors.standard.SplitContent.FRAGMENT_ID;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestUnpackContent {
@@ -60,6 +68,23 @@ public class TestUnpackContent {
private static final DateTimeFormatter TIMESTAMP_FORMATTER =
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssZ");
+ private static final String ZIP_ENTRY_OK = "ok.txt";
+ private static final String ZIP_ENTRY_CORRUPT = "corrupt.txt";
+ private static final String ZIP_ENTRY_FIRST = "first.txt";
+ private static final String ZIP_ENTRY_SECOND = "second.txt";
+ private static final String ZIP_ENTRY_KEEP = "keep.txt";
+ private static final String ZIP_ENTRY_SKIP = "skip.txt";
+ private static final String ZIP_ENTRY_KEEP_CONTENT = "keep-me";
+ private static final String ZIP_STORED_CRC_MATCH_CONTENT =
"payload-for-crc-match";
+ private static final String ZIP_STORED_CRC_MISMATCH_CONTENT =
"payload-for-crc-mismatch";
+ private static final String ZIP_DEFLATED_CRC_MATCH_CONTENT =
"deflated-payload-for-crc-match";
+ private static final String ZIP_DEFLATED_CRC_MISMATCH_CONTENT =
"deflated-payload-for-crc-mismatch";
+ private static final String ZIP_ENCRYPTION_PASSWORD =
String.class.getSimpleName();
+ private static final String ZIP_ENCRYPTION_CONTENTS =
TestRunner.class.getCanonicalName();
+ private static final byte[] ZIP_LOCAL_FILE_HEADER_SIGNATURE = {0x50, 0x4b,
0x03, 0x04};
+ private static final byte[] ZIP_CENTRAL_DIRECTORY_SIGNATURE = {0x50, 0x4b,
0x01, 0x02};
+ private static final byte[] ZIP_DATA_DESCRIPTOR_SIGNATURE = {0x50, 0x4b,
0x07, 0x08};
+
private final TestRunner runner = TestRunners.newTestRunner(new
UnpackContent());
private final TestRunner autoUnpackRunner = TestRunners.newTestRunner(new
UnpackContent());
@@ -249,6 +274,153 @@ public class TestUnpackContent {
flowFile.assertContentEquals(path.toFile());
}
}
+
+ @Test
+ public void testZipInvalidCrcRoutesToFailure() throws IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+
+ final byte[] zipBytes = createStoredZip(true,
Map.entry(ZIP_ENTRY_CORRUPT, ZIP_STORED_CRC_MISMATCH_CONTENT));
+ runner.enqueue(zipBytes);
+ runner.run();
+
+ assertOriginalRoutedToFailureOnly(zipBytes);
+ assertCrcMismatchLoggedWithoutStackTrace();
+ }
+
+ @Test
+ public void testZipStoredCrcValidRoutesToSuccess() throws IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+
+ final byte[] payload =
ZIP_STORED_CRC_MATCH_CONTENT.getBytes(StandardCharsets.UTF_8);
+ runner.enqueue(createStoredZip(false, Map.entry(ZIP_ENTRY_OK,
ZIP_STORED_CRC_MATCH_CONTENT)));
+ runner.run();
+
+ runner.assertTransferCount(UnpackContent.REL_FAILURE, 0);
+ runner.assertTransferCount(UnpackContent.REL_SUCCESS, 1);
+ runner.assertTransferCount(UnpackContent.REL_ORIGINAL, 1);
+
runner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS).getFirst().assertContentEquals(payload);
+ }
+
+ @Test
+ public void testZipDeflatedInvalidCrcRoutesToFailure() throws IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+
+ final byte[] zipBytes = createDeflatedZip(true,
Map.entry(ZIP_ENTRY_CORRUPT, ZIP_DEFLATED_CRC_MISMATCH_CONTENT));
+ runner.enqueue(zipBytes);
+ runner.run();
+
+ assertOriginalRoutedToFailureOnly(zipBytes);
+ assertCrcMismatchLoggedWithoutStackTrace();
+ }
+
+ @Test
+ public void testZipDeflatedCrcValidRoutesToSuccess() throws IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+
+ final String contents = ZIP_DEFLATED_CRC_MATCH_CONTENT;
+ runner.enqueue(createDeflatedZip(false, Map.entry(ZIP_ENTRY_OK,
contents)));
+ runner.run();
+
+ runner.assertTransferCount(UnpackContent.REL_FAILURE, 0);
+ runner.assertTransferCount(UnpackContent.REL_SUCCESS, 1);
+ runner.assertTransferCount(UnpackContent.REL_ORIGINAL, 1);
+
runner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS).getFirst()
+
.assertContentEquals(contents.getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testZipDeflatedInvalidCrcOnSecondEntryRoutesToFailure() throws
IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+
+ final byte[] zipBytes = createDeflatedZip(true,
+ Map.entry(ZIP_ENTRY_FIRST, "first-entry-valid-crc"),
+ Map.entry(ZIP_ENTRY_SECOND, "second-entry-invalid-crc"));
+ runner.enqueue(zipBytes);
+ runner.run();
+
+ assertOriginalRoutedToFailureOnly(zipBytes);
+ assertCrcMismatchLoggedWithoutStackTrace();
+ }
+
+ @Test
+ public void testZipStoredInvalidCrcOnSecondEntryRoutesToFailure() throws
IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+
+ final byte[] zipBytes = createStoredZip(true,
+ Map.entry(ZIP_ENTRY_FIRST, "first-stored-valid-crc"),
+ Map.entry(ZIP_ENTRY_SECOND, "second-stored-invalid-crc"));
+ runner.enqueue(zipBytes);
+ runner.run();
+
+ assertOriginalRoutedToFailureOnly(zipBytes);
+ assertCrcMismatchLoggedWithoutStackTrace();
+ }
+
+ @Test
+ public void testZipFileFilterExtractsMatchingEntryWhenCrcIsValid() throws
IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+ runner.setProperty(UnpackContent.FILE_FILTER, "^" + ZIP_ENTRY_KEEP +
"$");
+
+ runner.enqueue(createDeflatedZip(false,
+ Map.entry(ZIP_ENTRY_KEEP, ZIP_ENTRY_KEEP_CONTENT),
+ Map.entry(ZIP_ENTRY_SKIP, "skip-me")));
+ runner.run();
+
+ runner.assertTransferCount(UnpackContent.REL_FAILURE, 0);
+ runner.assertTransferCount(UnpackContent.REL_SUCCESS, 1);
+ runner.assertTransferCount(UnpackContent.REL_ORIGINAL, 1);
+
runner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS).getFirst()
+
.assertContentEquals(ZIP_ENTRY_KEEP_CONTENT.getBytes(StandardCharsets.UTF_8));
+
runner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS).getFirst()
+ .assertAttributeEquals(CoreAttributes.FILENAME.key(),
ZIP_ENTRY_KEEP);
+ }
+
+ @Test
+ public void testZipFileFilterStillFailsWhenSkippedEntryHasInvalidCrc()
throws IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+ runner.setProperty(UnpackContent.FILE_FILTER, "^" + ZIP_ENTRY_KEEP +
"$");
+
+ final byte[] zipBytes = createDeflatedZip(true,
+ Map.entry(ZIP_ENTRY_KEEP, ZIP_ENTRY_KEEP_CONTENT),
+ Map.entry(ZIP_ENTRY_SKIP, "corrupt-skipped-entry"));
+ runner.enqueue(zipBytes);
+ runner.run();
+
+ assertOriginalRoutedToFailureOnly(zipBytes);
+ assertCrcMismatchLoggedWithoutStackTrace();
+ }
+
+ @Test
+ public void testZipFileFilterFailsWhenOnlySkippedEntryHasInvalidCrc()
throws IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+ runner.setProperty(UnpackContent.FILE_FILTER, "^" + ZIP_ENTRY_KEEP +
"$");
+
+ final byte[] zipBytes = createStoredZip(true,
Map.entry(ZIP_ENTRY_SKIP, "not-extracted-and-corrupt"));
+ runner.enqueue(zipBytes);
+ runner.run();
+
+ assertOriginalRoutedToFailureOnly(zipBytes);
+ assertCrcMismatchLoggedWithoutStackTrace();
+ }
+
+ @Test
+ public void testZipDirectoryEntryAndFileWithValidCrcRoutesToSuccess()
throws IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
+
+ runner.enqueue(createDeflatedZip(false,
+ Map.entry("folder/", ""),
+ Map.entry("folder/file.txt", "nested-file")));
+ runner.run();
+
+ runner.assertTransferCount(UnpackContent.REL_FAILURE, 0);
+ runner.assertTransferCount(UnpackContent.REL_SUCCESS, 1);
+ runner.assertTransferCount(UnpackContent.REL_ORIGINAL, 1);
+
runner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS).getFirst()
+
.assertContentEquals("nested-file".getBytes(StandardCharsets.UTF_8));
+
runner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS).getFirst()
+ .assertAttributeEquals(CoreAttributes.FILENAME.key(),
"file.txt");
+ }
+
@Test
public void testZipEncodingField() {
runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
@@ -305,11 +477,10 @@ public class TestUnpackContent {
autoUnpackRunner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
autoUnpackRunner.setProperty(UnpackContent.ALLOW_STORED_ENTRIES_WITH_DATA_DESCRIPTOR,
"false");
autoUnpackRunner.setProperty(UnpackContent.ZIP_FILENAME_CHARSET,
"Cp437");
- final String password = String.class.getSimpleName();
- autoUnpackRunner.setProperty(UnpackContent.PASSWORD, password);
+ autoUnpackRunner.setProperty(UnpackContent.PASSWORD,
ZIP_ENCRYPTION_PASSWORD);
- final char[] streamPassword = password.toCharArray();
- final String contents = TestRunner.class.getCanonicalName();
+ final char[] streamPassword = ZIP_ENCRYPTION_PASSWORD.toCharArray();
+ final String contents = ZIP_ENCRYPTION_CONTENTS;
String specialChar = "\u00E4";
String pathInZip = "path_with_special_%s_char/".formatted(specialChar);
String filename =
"filename_with_special_char%s.txt".formatted(specialChar);
@@ -346,9 +517,8 @@ public class TestUnpackContent {
public void testZipEncryptionNoPasswordConfigured() throws IOException {
autoUnpackRunner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
- final String password = String.class.getSimpleName();
- final char[] streamPassword = password.toCharArray();
- final String contents = TestRunner.class.getCanonicalName();
+ final char[] streamPassword = ZIP_ENCRYPTION_PASSWORD.toCharArray();
+ final String contents = ZIP_ENCRYPTION_CONTENTS;
final byte[] zipEncrypted = createZipEncrypted(EncryptionMethod.AES,
streamPassword, contents);
autoUnpackRunner.enqueue(zipEncrypted);
@@ -656,11 +826,10 @@ public class TestUnpackContent {
private void runZipEncryptionMethod(final EncryptionMethod
encryptionMethod) throws IOException {
runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.ZIP_FORMAT);
runner.setProperty(UnpackContent.ALLOW_STORED_ENTRIES_WITH_DATA_DESCRIPTOR,
"false");
- final String password = String.class.getSimpleName();
- runner.setProperty(UnpackContent.PASSWORD, password);
+ runner.setProperty(UnpackContent.PASSWORD, ZIP_ENCRYPTION_PASSWORD);
- final char[] streamPassword = password.toCharArray();
- final String contents = TestRunner.class.getCanonicalName();
+ final char[] streamPassword = ZIP_ENCRYPTION_PASSWORD.toCharArray();
+ final String contents = ZIP_ENCRYPTION_CONTENTS;
final byte[] zipEncrypted = createZipEncrypted(encryptionMethod,
streamPassword, contents);
runner.enqueue(zipEncrypted);
@@ -711,4 +880,124 @@ public class TestUnpackContent {
return outputStream.toByteArray();
}
+
+ private void assertOriginalRoutedToFailureOnly(final byte[] originalZip)
throws IOException {
+ runner.assertTransferCount(UnpackContent.REL_FAILURE, 1);
+ runner.assertTransferCount(UnpackContent.REL_SUCCESS, 0);
+ runner.assertTransferCount(UnpackContent.REL_ORIGINAL, 0);
+
runner.getFlowFilesForRelationship(UnpackContent.REL_FAILURE).getFirst().assertContentEquals(originalZip);
+ }
+
+ private void assertCrcMismatchLoggedWithoutStackTrace() {
+ final List<LogMessage> errors = runner.getLogger().getErrorMessages();
+ assertFalse(errors.isEmpty(), "Expected an error log for CRC
mismatch");
+ final LogMessage error = errors.getFirst();
+ final String details = error.getMsg() +
Arrays.toString(error.getArgs());
+ assertTrue(details.contains("CRC mismatch"), "Error log should
describe the CRC mismatch: " + details);
+ assertNull(error.getThrowable(), "CRC mismatch should be logged
without a stack trace");
+ }
+
+ /**
+ * Builds a STORED zip in memory. When {@code invertLastEntryCrc} is true,
the CRC-32 of the last
+ * entry in the local file header and central directory is
bitwise-inverted so the archive is
+ * well-formed but the checksum no longer matches the entry bytes.
+ */
+ @SafeVarargs
+ private static byte[] createStoredZip(final boolean invertLastEntryCrc,
final Map.Entry<String, String>... entries) throws IOException {
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try (java.util.zip.ZipOutputStream zipOutputStream = new
java.util.zip.ZipOutputStream(outputStream)) {
+ zipOutputStream.setMethod(java.util.zip.ZipOutputStream.STORED);
+ for (final Map.Entry<String, String> entry : entries) {
+ final byte[] payload =
entry.getValue().getBytes(StandardCharsets.UTF_8);
+ final CRC32 crc32 = new CRC32();
+ crc32.update(payload);
+ final ZipEntry zipEntry = new ZipEntry(entry.getKey());
+ zipEntry.setMethod(ZipEntry.STORED);
+ zipEntry.setCrc(crc32.getValue());
+ zipEntry.setSize(payload.length);
+ zipEntry.setCompressedSize(payload.length);
+ zipOutputStream.putNextEntry(zipEntry);
+ zipOutputStream.write(payload);
+ zipOutputStream.closeEntry();
+ }
+ }
+
+ final byte[] zipBytes = outputStream.toByteArray();
+ if (!invertLastEntryCrc) {
+ return zipBytes;
+ }
+
+ final int centralDirectoryIndex = indexOf(zipBytes,
ZIP_CENTRAL_DIRECTORY_SIGNATURE);
+ assertTrue(centralDirectoryIndex > 0, "ZIP central directory signature
not found");
+ corruptCrcAt(zipBytes, lastIndexOf(zipBytes,
ZIP_LOCAL_FILE_HEADER_SIGNATURE, centralDirectoryIndex) + 14);
+ corruptCrcAt(zipBytes, lastIndexOf(zipBytes,
ZIP_CENTRAL_DIRECTORY_SIGNATURE, zipBytes.length) + 16);
+ return zipBytes;
+ }
+
+ /**
+ * Builds a DEFLATED zip (Java default: data descriptor after the
compressed bytes). When
+ * {@code invertLastEntryCrc} is true, the CRC in that last entry's data
descriptor and central
+ * directory is inverted.
+ */
+ @SafeVarargs
+ private static byte[] createDeflatedZip(final boolean invertLastEntryCrc,
final Map.Entry<String, String>... entries) throws IOException {
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try (java.util.zip.ZipOutputStream zipOutputStream = new
java.util.zip.ZipOutputStream(outputStream)) {
+ for (final Map.Entry<String, String> entry : entries) {
+ final ZipEntry zipEntry = new ZipEntry(entry.getKey());
+ zipEntry.setMethod(ZipEntry.DEFLATED);
+ zipOutputStream.putNextEntry(zipEntry);
+
zipOutputStream.write(entry.getValue().getBytes(StandardCharsets.UTF_8));
+ zipOutputStream.closeEntry();
+ }
+ }
+
+ final byte[] zipBytes = outputStream.toByteArray();
+ if (!invertLastEntryCrc) {
+ return zipBytes;
+ }
+
+ final int centralDirectoryIndex = indexOf(zipBytes,
ZIP_CENTRAL_DIRECTORY_SIGNATURE);
+ assertTrue(centralDirectoryIndex > 0, "ZIP central directory signature
not found");
+ final int dataDescriptorIndex = lastIndexOf(zipBytes,
ZIP_DATA_DESCRIPTOR_SIGNATURE, centralDirectoryIndex);
+ assertTrue(dataDescriptorIndex >= 0, "ZIP data descriptor signature
not found");
+ corruptCrcAt(zipBytes, dataDescriptorIndex + 4);
+ corruptCrcAt(zipBytes, lastIndexOf(zipBytes,
ZIP_CENTRAL_DIRECTORY_SIGNATURE, zipBytes.length) + 16);
+ return zipBytes;
+ }
+
+ private static void corruptCrcAt(final byte[] zipBytes, final int
crcIndex) {
+ final int storedCrc = ByteBuffer.wrap(zipBytes, crcIndex,
4).order(ByteOrder.LITTLE_ENDIAN).getInt();
+ ByteBuffer.wrap(zipBytes, crcIndex,
4).order(ByteOrder.LITTLE_ENDIAN).putInt(storedCrc ^ 0xffffffff);
+ }
+
+ private static int indexOf(final byte[] haystack, final byte[] needle) {
+ return indexOf(haystack, needle, 0, haystack.length);
+ }
+
+ private static int lastIndexOf(final byte[] haystack, final byte[] needle,
final int endExclusive) {
+ outer:
+ for (int i = endExclusive - needle.length; i >= 0; i--) {
+ for (int j = 0; j < needle.length; j++) {
+ if (haystack[i + j] != needle[j]) {
+ continue outer;
+ }
+ }
+ return i;
+ }
+ return -1;
+ }
+
+ private static int indexOf(final byte[] haystack, final byte[] needle,
final int start, final int endExclusive) {
+ outer:
+ for (int i = start; i <= endExclusive - needle.length; i++) {
+ for (int j = 0; j < needle.length; j++) {
+ if (haystack[i + j] != needle[j]) {
+ continue outer;
+ }
+ }
+ return i;
+ }
+ return -1;
+ }
}
