diff --git a/phase4-lib/src/main/java/com/helger/phase4/attachment/WSS4JAttachment.java b/phase4-lib/src/main/java/com/helger/phase4/attachment/WSS4JAttachment.java index f0f5bda1e..313613872 100644 --- a/phase4-lib/src/main/java/com/helger/phase4/attachment/WSS4JAttachment.java +++ b/phase4-lib/src/main/java/com/helger/phase4/attachment/WSS4JAttachment.java @@ -54,6 +54,7 @@ import com.helger.phase4.logging.Phase4LoggerFactory; import com.helger.phase4.model.message.MessageHelperMethods; import com.helger.phase4.util.AS4ResourceHelper; +import com.helger.phase4.util.MarkableFileInputStream; import jakarta.activation.DataHandler; import jakarta.activation.DataSource; @@ -462,8 +463,8 @@ public static WSS4JAttachment createOutgoingFileAttachment (@NonNull final File } // Set a stream provider that can be read multiple times (opens a new - // FileInputStream internally) - final IHasInputStream aISP = HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aRealFile)); + // stream internally) + final IHasInputStream aISP = HasInputStream.multiple (() -> MarkableFileInputStream.create (aRealFile)); ret.setSourceStreamProvider (aISP); if (eCompressionMode != null) { @@ -531,7 +532,7 @@ public static WSS4JAttachment createOutgoingFileAttachment (final byte @NonNull aOS.write (aSrcData); } } - final IHasInputStream aISP = HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aRealFile)); + final IHasInputStream aISP = HasInputStream.multiple (() -> MarkableFileInputStream.create (aRealFile)); ret.setSourceStreamProvider (aISP); // Preserve the compressed data for non-repudiation purposes - the // signature digests are calculated over the compressed data @@ -629,7 +630,7 @@ public static WSS4JAttachment createIncomingFileAttachment (@NonNull final AS4In aOS.write (nProbedByte); StreamHelper.copyInputStreamToOutputStream (aDecodedIS, aOS); } - ret.setSourceStreamProvider (HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aTempFile))); + ret.setSourceStreamProvider (HasInputStream.multiple (() -> MarkableFileInputStream.create (aTempFile))); } // Read all MIME part headers @@ -739,7 +740,7 @@ public static WSS4JAttachment createIncomingFileAttachment (@NonNull final MimeB { aBodyPart.getDataHandler ().writeTo (aOS); } - ret.setSourceStreamProvider (HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aTempFile))); + ret.setSourceStreamProvider (HasInputStream.multiple (() -> MarkableFileInputStream.create (aTempFile))); } // Read all MIME part headers diff --git a/phase4-lib/src/main/java/com/helger/phase4/incoming/AS4IncomingHandler.java b/phase4-lib/src/main/java/com/helger/phase4/incoming/AS4IncomingHandler.java index 22e9cc591..28298ef9b 100644 --- a/phase4-lib/src/main/java/com/helger/phase4/incoming/AS4IncomingHandler.java +++ b/phase4-lib/src/main/java/com/helger/phase4/incoming/AS4IncomingHandler.java @@ -98,6 +98,7 @@ import com.helger.phase4.profile.IAS4ProfileValidator.EAS4ProfileValidationMode; import com.helger.phase4.util.AS4ResourceHelper; import com.helger.phase4.util.AS4XMLHelper; +import com.helger.phase4.util.MarkableFileInputStream; import com.helger.phase4.util.Phase4Exception; import com.helger.phase4.util.Phase4IncomingException; import com.helger.web.multipart.MultipartProgressNotifier; @@ -651,7 +652,7 @@ private static IHasInputStream _createReadMultipleISP (@NonNull final AS4Resourc } aTempFileWrapper.set (aTempFile); } - return FileHelper.getBufferedInputStream (aTempFile); + return MarkableFileInputStream.create (aTempFile); } catch (final IOException ex) { diff --git a/phase4-lib/src/main/java/com/helger/phase4/incoming/soap/SoapHeaderElementProcessorWSS4J.java b/phase4-lib/src/main/java/com/helger/phase4/incoming/soap/SoapHeaderElementProcessorWSS4J.java index a6a8bc9b8..70753a123 100644 --- a/phase4-lib/src/main/java/com/helger/phase4/incoming/soap/SoapHeaderElementProcessorWSS4J.java +++ b/phase4-lib/src/main/java/com/helger/phase4/incoming/soap/SoapHeaderElementProcessorWSS4J.java @@ -70,6 +70,7 @@ import com.helger.phase4.model.error.EEbmsError; import com.helger.phase4.model.pmode.IPMode; import com.helger.phase4.model.pmode.leg.PModeLeg; +import com.helger.phase4.util.MarkableFileInputStream; import com.helger.phase4.wss.WSSConfigManager; import com.helger.phase4.wss.WSSSynchronizer; import com.helger.xml.XMLHelper; @@ -356,7 +357,7 @@ private ESuccess _verifyAndDecrypt (@NonNull final Document aSOAPDoc, { LOGGER.error ("Failed to write response attachment to temporary file '" + aTempFile.getAbsolutePath () + "'"); } - aResponseAttachment.setSourceStreamProvider (HasInputStream.multiple (() -> FileHelper.getBufferedInputStream (aTempFile))); + aResponseAttachment.setSourceStreamProvider (HasInputStream.multiple (() -> MarkableFileInputStream.create (aTempFile))); } // Remember in State diff --git a/phase4-lib/src/main/java/com/helger/phase4/util/MarkableFileInputStream.java b/phase4-lib/src/main/java/com/helger/phase4/util/MarkableFileInputStream.java new file mode 100644 index 000000000..94c1538c9 --- /dev/null +++ b/phase4-lib/src/main/java/com/helger/phase4/util/MarkableFileInputStream.java @@ -0,0 +1,242 @@ +/* + * Copyright (C) 2015-2026 Philip Helger (www.helger.com) + * philip[at]helger[dot]com + * + * Licensed 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 com.helger.phase4.util; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.StandardOpenOption; +import java.util.Objects; + +import org.jspecify.annotations.NonNull; +import org.jspecify.annotations.Nullable; +import org.slf4j.Logger; + +import com.helger.annotation.concurrent.NotThreadSafe; +import com.helger.base.enforce.ValueEnforcer; +import com.helger.phase4.logging.Phase4LoggerFactory; + +/** + * A buffered {@link InputStream} on a {@link File} that supports {@link #mark(int)} and + * {@link #reset()} with constant heap usage, by re-positioning the underlying {@link FileChannel} + * instead of buffering all the bytes read after the mark.
+ * This is relevant for the streams handed over to WSS4J: for every signed attachment, + * AttachmentContentSignatureTransform#processAttachment calls + * mark (Integer.MAX_VALUE) on the source stream, reads it to the end to calculate the + * digest and calls reset () afterwards, so that the attachment stays readable. On a + * heap buffering stream (like BufferedInputStream or + * NonBlockingBufferedInputStream) that mark/read/reset sequence keeps the complete + * attachment on the heap, so the heap usage scales with the attachment size. With this class it + * stays constant, and the 2 GB limit of heap buffering streams does not apply. See issue #380.
+ * The mark position is initially 0, so a {@link #reset()} without a preceding {@link #mark(int)} + * re-reads from the beginning (like {@link java.io.ByteArrayInputStream}). + * + * @since 4.6.0 + */ +@NotThreadSafe +public class MarkableFileInputStream extends InputStream +{ + /** The default size of the internal read buffer in bytes */ + public static final int DEFAULT_BUFFER_SIZE = 16 * 1024; + + private static final Logger LOGGER = Phase4LoggerFactory.getLogger (MarkableFileInputStream.class); + + private final FileChannel m_aChannel; + // Read buffer; between reads always in "drain" mode: position..limit are the + // unread bytes + private final ByteBuffer m_aBuffer; + // File offset of the next byte to be read from the channel (mirrors + // FileChannel.position) + private long m_nChannelPos = 0; + // File offset to fall back to on reset + private long m_nMarkPos = 0; + + /** + * Constructor using {@link #DEFAULT_BUFFER_SIZE}. + * + * @param aFile + * The file to read. May not be null. + * @throws IOException + * If the file cannot be opened for reading. + */ + public MarkableFileInputStream (@NonNull final File aFile) throws IOException + { + this (aFile, DEFAULT_BUFFER_SIZE); + } + + /** + * Constructor. + * + * @param aFile + * The file to read. May not be null. + * @param nBufferSize + * The size of the internal read buffer in bytes. Must be > 0. + * @throws IOException + * If the file cannot be opened for reading. + */ + public MarkableFileInputStream (@NonNull final File aFile, final int nBufferSize) throws IOException + { + ValueEnforcer.notNull (aFile, "File"); + ValueEnforcer.isGT0 (nBufferSize, "BufferSize"); + m_aChannel = FileChannel.open (aFile.toPath (), StandardOpenOption.READ); + m_aBuffer = ByteBuffer.allocate (nBufferSize); + // The buffer starts out empty + m_aBuffer.limit (0); + } + + /** + * @return The offset of the next byte to be delivered by this stream. + */ + private long _getLogicalPos () + { + return m_nChannelPos - m_aBuffer.remaining (); + } + + /** + * Fill the internal buffer from the channel. + * + * @return false if EOF was reached. + */ + private boolean _fill () throws IOException + { + m_aBuffer.clear (); + final int nRead = m_aChannel.read (m_aBuffer); + m_aBuffer.flip (); + if (nRead <= 0) + return false; + m_nChannelPos += nRead; + return true; + } + + /** + * Re-position the channel and discard all buffered bytes. + */ + private void _seek (final long nPos) throws IOException + { + m_aChannel.position (nPos); + m_nChannelPos = nPos; + m_aBuffer.position (0).limit (0); + } + + @Override + public int read () throws IOException + { + if (!m_aBuffer.hasRemaining () && !_fill ()) + return -1; + return m_aBuffer.get () & 0xff; + } + + @Override + public int read (final byte @NonNull [] aBuf, final int nOfs, final int nLen) throws IOException + { + Objects.checkFromIndexSize (nOfs, nLen, aBuf.length); + if (nLen == 0) + return 0; + + if (!m_aBuffer.hasRemaining ()) + { + // Read large requests directly from the channel, bypassing the buffer + if (nLen >= m_aBuffer.capacity ()) + { + final int nRead = m_aChannel.read (ByteBuffer.wrap (aBuf, nOfs, nLen)); + if (nRead > 0) + m_nChannelPos += nRead; + return nRead; + } + if (!_fill ()) + return -1; + } + final int nRead = Math.min (m_aBuffer.remaining (), nLen); + m_aBuffer.get (aBuf, nOfs, nRead); + return nRead; + } + + @Override + public long skip (final long nBytes) throws IOException + { + if (nBytes <= 0) + return 0; + final long nCurPos = _getLogicalPos (); + final long nSize = m_aChannel.size (); + if (nCurPos >= nSize) + return 0; + // Never skip beyond EOF + final long nSkipped = Math.min (nBytes, nSize - nCurPos); + _seek (nCurPos + nSkipped); + return nSkipped; + } + + @Override + public int available () throws IOException + { + final long nRemaining = m_aBuffer.remaining () + Math.max (0, m_aChannel.size () - m_nChannelPos); + return nRemaining > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) nRemaining; + } + + @Override + public boolean markSupported () + { + return true; + } + + /** + * {@inheritDoc}
+ * The read limit parameter is ignored: as {@link #reset()} only re-positions the underlying + * {@link FileChannel}, the mark can never be invalidated, no matter how many bytes are read. + */ + @Override + public void mark (final int nReadLimit) + { + m_nMarkPos = _getLogicalPos (); + } + + @Override + public void reset () throws IOException + { + _seek (m_nMarkPos); + } + + @Override + public void close () throws IOException + { + m_aChannel.close (); + } + + /** + * Factory method that logs an error and returns null if the file cannot be opened - + * same semantics as FileHelper.getBufferedInputStream. + * + * @param aFile + * The file to read. May not be null. + * @return null if the file cannot be opened for reading. + */ + @Nullable + public static MarkableFileInputStream create (@NonNull final File aFile) + { + try + { + return new MarkableFileInputStream (aFile); + } + catch (final IOException ex) + { + LOGGER.warn ("Failed to open file '" + aFile.getAbsolutePath () + "' for reading", ex); + return null; + } + } +} diff --git a/phase4-lib/src/test/java/com/helger/phase4/attachment/WSS4JAttachmentTest.java b/phase4-lib/src/test/java/com/helger/phase4/attachment/WSS4JAttachmentTest.java index e40874f6c..2cfa8b095 100644 --- a/phase4-lib/src/test/java/com/helger/phase4/attachment/WSS4JAttachmentTest.java +++ b/phase4-lib/src/test/java/com/helger/phase4/attachment/WSS4JAttachmentTest.java @@ -26,6 +26,7 @@ import java.io.File; import java.io.IOException; +import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.util.concurrent.ThreadLocalRandom; @@ -40,6 +41,7 @@ import com.helger.io.file.SimpleFileIO; import com.helger.mime.CMimeType; import com.helger.phase4.util.AS4ResourceHelper; +import com.helger.phase4.util.MarkableFileInputStream; import jakarta.mail.MessagingException; @@ -257,6 +259,60 @@ public void testIncomingEmpty () throws IOException, MessagingException } } + private static void _assertMarkableAndReReadable (@NonNull final WSS4JAttachment a) throws IOException + { + final InputStream aIS = a.getSourceStream (); + // Anything else buffers the whole attachment on the heap on mark/reset + assertTrue ("Not a " + MarkableFileInputStream.class.getSimpleName () + ": " + aIS, + aIS instanceof MarkableFileInputStream); + + // Consume the stream the way WSS4J digests a signed attachment + aIS.mark (Integer.MAX_VALUE); + final byte [] aDigested = aIS.readAllBytes (); + aIS.reset (); + + // Afterwards WSS4J re-uses the same stream object for the attachment + // content + assertArrayEquals (aDigested, aIS.readAllBytes ()); + } + + /** + * All file backed attachment source streams must support mark/reset with constant heap usage, + * because WSS4J digests each signed attachment via mark/read-to-end/reset. See issue #380. + */ + @Test + public void testFileBackedSourceStreamsAreMarkable () throws IOException, MessagingException + { + // Larger than the in-memory threshold, so that a temporary file is used + final byte [] aContent = new byte [WSS4JAttachment.MAX_IN_MEMORY_BYTES + 1024]; + ThreadLocalRandom.current ().nextBytes (aContent); + + final File fSrc = m_aRule.newFile ("payload.bin"); + SimpleFileIO.writeFile (fSrc, aContent); + + try (final AS4ResourceHelper aResHelper = new AS4ResourceHelper ()) + { + // Outgoing, uncompressed - the source file is used as-is + final AS4OutgoingAttachment aOA1 = AS4OutgoingAttachment.builder () + .data (fSrc) + .mimeType (CMimeType.APPLICATION_OCTET_STREAM) + .build (); + _assertMarkableAndReReadable (WSS4JAttachment.createOutgoingFileAttachment (aOA1, aResHelper)); + + // Outgoing, compressed - a temporary file with the compressed content + final AS4OutgoingAttachment aOA2 = AS4OutgoingAttachment.builder () + .data (aContent) + .mimeType (CMimeType.APPLICATION_OCTET_STREAM) + .compressionGZIP () + .build (); + _assertMarkableAndReReadable (WSS4JAttachment.createOutgoingFileAttachment (aOA2, aResHelper)); + + // Incoming, above the in-memory threshold - a temporary file + final AS4IncomingMimePart aMimePart = _createIncomingMimePart (aContent); + _assertMarkableAndReReadable (WSS4JAttachment.createIncomingFileAttachment (aMimePart, aResHelper)); + } + } + @Test public void testIncomingExactlyThresholdSize () throws IOException, MessagingException { diff --git a/phase4-lib/src/test/java/com/helger/phase4/util/MarkableFileInputStreamTest.java b/phase4-lib/src/test/java/com/helger/phase4/util/MarkableFileInputStreamTest.java new file mode 100644 index 000000000..c2cf3c873 --- /dev/null +++ b/phase4-lib/src/test/java/com/helger/phase4/util/MarkableFileInputStreamTest.java @@ -0,0 +1,231 @@ +/* + * Copyright (C) 2015-2026 Philip Helger (www.helger.com) + * philip[at]helger[dot]com + * + * Licensed 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 com.helger.phase4.util; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.util.Arrays; +import java.util.Random; + +import org.jspecify.annotations.NonNull; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import com.helger.base.io.nonblocking.NonBlockingByteArrayOutputStream; +import com.helger.io.file.SimpleFileIO; + +/** + * Test class for class {@link MarkableFileInputStream}. + */ +public final class MarkableFileInputStreamTest +{ + @Rule + public final TemporaryFolder m_aRule = new TemporaryFolder (); + + @NonNull + private static byte [] _createRandomBytes (final int nLen) + { + final byte [] ret = new byte [nLen]; + // Deterministic seed + new Random (20260803L).nextBytes (ret); + return ret; + } + + @NonNull + private File _createFile (final byte [] aContent) throws IOException + { + final File ret = m_aRule.newFile (); + SimpleFileIO.writeFile (ret, aContent); + return ret; + } + + /** + * Consume the stream exactly the way WSS4J's + * AttachmentContentSignatureTransform#processAttachment does when calculating the + * digest of a signed attachment: mark (Integer.MAX_VALUE), read to the end, + * reset (). Afterwards WSS4J re-uses the same stream object for the attachment + * content, so a subsequent full read must deliver the identical bytes from the start. + */ + @NonNull + private static byte [] _digestLikeWSS4J (@NonNull final InputStream aIS) throws IOException + { + assertTrue ("WSS4J relies on mark/reset support", aIS.markSupported ()); + aIS.mark (Integer.MAX_VALUE); + final byte [] ret = _readAll (aIS); + aIS.reset (); + return ret; + } + + @NonNull + private static byte [] _readAll (@NonNull final InputStream aIS) throws IOException + { + try (final NonBlockingByteArrayOutputStream aAll = new NonBlockingByteArrayOutputStream ()) + { + int nBytes; + // Same buffer size as WSS4J uses + final byte [] aBuf = new byte [8192]; + while ((nBytes = aIS.read (aBuf)) != -1) + aAll.write (aBuf, 0, nBytes); + return aAll.toByteArray (); + } + } + + @Test + public void testDigestThenReRead () throws IOException + { + // Clearly larger than the internal buffer + final byte [] aPayload = _createRandomBytes (1024 * 1024 + 17); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + assertArrayEquals ("The digested bytes must be the file content", aPayload, _digestLikeWSS4J (aIS)); + + // Re-read the same stream object from the beginning + assertArrayEquals ("The content after reset must be identical", aPayload, _readAll (aIS)); + } + } + + @Test + public void testMarkAtNonZeroOffset () throws IOException + { + final byte [] aPayload = _createRandomBytes (100_000); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + // Read some bytes, crossing the internal buffer boundary + final byte [] aHead = new byte [MarkableFileInputStream.DEFAULT_BUFFER_SIZE + 100]; + int nRead = 0; + while (nRead < aHead.length) + nRead += aIS.read (aHead, nRead, aHead.length - nRead); + assertArrayEquals (Arrays.copyOf (aPayload, aHead.length), aHead); + + aIS.mark (Integer.MAX_VALUE); + final byte [] aTail1 = _readAll (aIS); + aIS.reset (); + final byte [] aTail2 = _readAll (aIS); + assertArrayEquals (Arrays.copyOfRange (aPayload, aHead.length, aPayload.length), aTail1); + assertArrayEquals (aTail1, aTail2); + } + } + + @Test + public void testResetWithoutMarkReReadsFromStart () throws IOException + { + final byte [] aPayload = _createRandomBytes (1000); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + assertArrayEquals (aPayload, _readAll (aIS)); + aIS.reset (); + assertArrayEquals (aPayload, _readAll (aIS)); + } + } + + @Test + public void testSingleByteRead () throws IOException + { + // Small file, read byte by byte + final byte [] aPayload = _createRandomBytes (100); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f, 8)) + { + for (final byte b : aPayload) + assertEquals (b & 0xff, aIS.read ()); + assertEquals (-1, aIS.read ()); + } + } + + @Test + public void testEmptyFile () throws IOException + { + final File f = _createFile (new byte [0]); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + assertEquals (0, aIS.available ()); + assertEquals (-1, aIS.read ()); + aIS.mark (Integer.MAX_VALUE); + aIS.reset (); + assertEquals (-1, aIS.read ()); + } + } + + @Test + public void testSkipAndAvailable () throws IOException + { + final byte [] aPayload = _createRandomBytes (50_000); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f)) + { + assertEquals (aPayload.length, aIS.available ()); + assertEquals (0, aIS.skip (0)); + assertEquals (0, aIS.skip (-1)); + assertEquals (10_000, aIS.skip (10_000)); + assertEquals (aPayload.length - 10_000, aIS.available ()); + assertEquals (aPayload[10_000] & 0xff, aIS.read ()); + // Skipping beyond EOF is truncated + assertEquals (aPayload.length - 10_001, aIS.skip (Long.MAX_VALUE)); + assertEquals (0, aIS.skip (1)); + assertEquals (-1, aIS.read ()); + assertEquals (0, aIS.available ()); + } + } + + @Test + public void testReadLargerThanBuffer () throws IOException + { + // Requests larger than the internal buffer are served from the channel + // directly + final byte [] aPayload = _createRandomBytes (70_000); + final File f = _createFile (aPayload); + + try (final MarkableFileInputStream aIS = new MarkableFileInputStream (f, 1024)) + { + aIS.mark (Integer.MAX_VALUE); + final byte [] aBuf = new byte [aPayload.length]; + int nTotal = 0; + int nRead; + while (nTotal < aBuf.length && (nRead = aIS.read (aBuf, nTotal, aBuf.length - nTotal)) != -1) + nTotal += nRead; + assertEquals (aPayload.length, nTotal); + assertArrayEquals (aPayload, aBuf); + assertEquals (-1, aIS.read ()); + + // The mark must still be valid, no matter how many bytes were read + aIS.reset (); + assertArrayEquals (aPayload, _readAll (aIS)); + } + } + + @Test + public void testCreateNonExistingFile () + { + assertNull (MarkableFileInputStream.create (new File (m_aRule.getRoot (), "does-not-exist.bin"))); + } +}