exceptionfactory commented on code in PR #11636:
URL: https://github.com/apache/nifi/pull/11636#discussion_r3962298822
##########
nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java:
##########
@@ -509,24 +557,36 @@ protected boolean isFileEntryMatched(final boolean
directory, final String fileN
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 {
Review Comment:
Although this is a carry over from the previous implementation, each
parameter can be declared `final`.
##########
nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java:
##########
@@ -452,10 +490,14 @@ public void unpack(final ProcessSession session, final
FlowFile source, final Li
}
}
});
+ 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;
Review Comment:
This static value could be removed and
[ZipArchiveEntry.CRC_UNKNOWN](https://commons.apache.org/proper/commons-compress/apidocs/org/apache/commons/compress/archivers/zip/ZipArchiveEntry.html#CRC_UNKNOWN)
can be used.
##########
nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java:
##########
@@ -711,4 +866,128 @@ private byte[] createZipEncryptedCp437(final
EncryptionMethod encryptionMethod,
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};
Review Comment:
Recommend declaring these as static final variables.
##########
nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java:
##########
@@ -249,6 +257,153 @@ public void testInvalidZip() throws IOException {
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);
Review Comment:
A number of the names and values are reused across methods, which could be
moved to static final variables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
