This is an automated email from the ASF dual-hosted git repository. joewitt pushed a commit to branch NIFI-15718 in repository https://gitbox.apache.org/repos/asf/nifi.git
commit bd5e09cfd4c0c3d8378f4c544bc97e318a70e66e Author: Joseph Witt <[email protected]> AuthorDate: Sun Sep 6 16:37:03 2026 -0700 NIFI-15718 Validate ZIP CRC-32 checksums in UnpackContent Corrupt archives that declare a CRC and fail the check now route to failure instead of emitting extracted entries. --- .../nifi/processors/standard/UnpackContent.java | 163 +++++++++--- .../processors/standard/TestUnpackContent.java | 279 +++++++++++++++++++++ 2 files changed, 411 insertions(+), 31 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 f1a4c70dc38..8e46a0c035c 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,22 @@ 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.") + + "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 +179,8 @@ 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 +188,8 @@ 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 +227,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 +339,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 +391,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 +407,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,10 +490,14 @@ public class UnpackContent extends AbstractProcessor { } } }); + return UnpackResult.success(); } } private static class ZipUnpacker extends Unpacker { + // ZipArchiveEntry.getCrc() returns -1 when the archive does not declare a checksum for the entry + private static final long UNKNOWN_CRC = -1; + private final char[] password; private final boolean allowStoredEntriesWithDataDescriptor; private final Charset filenameEncoding; @@ -467,13 +509,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 +557,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, boolean directory, String zipEntryName, 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 +624,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 == UNKNOWN_CRC || 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 +661,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 +696,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 +758,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 +800,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..310e77d8acf 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 { @@ -249,6 +257,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("corrupt.txt", "payload-for-crc-mismatch")); + 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 = "payload-for-crc-match".getBytes(StandardCharsets.UTF_8); + runner.enqueue(createStoredZip(false, Map.entry("ok.txt", "payload-for-crc-match"))); + 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("corrupt.txt", "deflated-payload-for-crc-mismatch")); + 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 = "deflated-payload-for-crc-match"; + runner.enqueue(createDeflatedZip(false, Map.entry("ok.txt", 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("first.txt", "first-entry-valid-crc"), + Map.entry("second.txt", "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("first.txt", "first-stored-valid-crc"), + Map.entry("second.txt", "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, "^keep.txt$"); + + runner.enqueue(createDeflatedZip(false, + Map.entry("keep.txt", "keep-me"), + Map.entry("skip.txt", "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("keep-me".getBytes(StandardCharsets.UTF_8)); + runner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS).getFirst() + .assertAttributeEquals(CoreAttributes.FILENAME.key(), "keep.txt"); + } + + @Test + public void testZipFileFilterStillFailsWhenSkippedEntryHasInvalidCrc() throws IOException { + runner.setProperty(UnpackContent.PACKAGING_FORMAT, UnpackContent.PackageFormat.ZIP_FORMAT); + runner.setProperty(UnpackContent.FILE_FILTER, "^keep.txt$"); + + final byte[] zipBytes = createDeflatedZip(true, + Map.entry("keep.txt", "keep-me"), + Map.entry("skip.txt", "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, "^keep.txt$"); + + final byte[] zipBytes = createStoredZip(true, Map.entry("skip.txt", "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); @@ -711,4 +866,128 @@ 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 byte[] localFileHeaderSignature = {0x50, 0x4b, 0x03, 0x04}; + final byte[] centralDirectorySignature = {0x50, 0x4b, 0x01, 0x02}; + final int centralDirectoryIndex = indexOf(zipBytes, centralDirectorySignature); + assertTrue(centralDirectoryIndex > 0, "ZIP central directory signature not found"); + corruptCrcAt(zipBytes, lastIndexOf(zipBytes, localFileHeaderSignature, centralDirectoryIndex) + 14); + corruptCrcAt(zipBytes, lastIndexOf(zipBytes, centralDirectorySignature, 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 byte[] centralDirectorySignature = {0x50, 0x4b, 0x01, 0x02}; + final byte[] dataDescriptorSignature = {0x50, 0x4b, 0x07, 0x08}; + final int centralDirectoryIndex = indexOf(zipBytes, centralDirectorySignature); + assertTrue(centralDirectoryIndex > 0, "ZIP central directory signature not found"); + final int dataDescriptorIndex = lastIndexOf(zipBytes, dataDescriptorSignature, centralDirectoryIndex); + assertTrue(dataDescriptorIndex >= 0, "ZIP data descriptor signature not found"); + corruptCrcAt(zipBytes, dataDescriptorIndex + 4); + corruptCrcAt(zipBytes, lastIndexOf(zipBytes, centralDirectorySignature, 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; + } }
