Bound heap usage when digesting signed attachments

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

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)
       {

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

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.<br>
+ * This is relevant for the streams handed over to WSS4J: for every signed attachment,
+ * <code>AttachmentContentSignatureTransform#processAttachment</code> calls
+ * <code>mark (Integer.MAX_VALUE)</code> on the source stream, reads it to the end to calculate the
+ * digest and calls <code>reset ()</code> afterwards, so that the attachment stays readable. On a
+ * heap buffering stream (like <code>BufferedInputStream</code> or
+ * <code>NonBlockingBufferedInputStream</code>) 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.<br>
+ * 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 <code>null</code>.
+   * @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 <code>null</code>.
+   * @param nBufferSize
+   *        The size of the internal read buffer in bytes. Must be &gt; 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 <code>false</code> 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}<br>
+   * 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 <code>null</code> if the file cannot be opened -
+   * same semantics as <code>FileHelper.getBufferedInputStream</code>.
+   *
+   * @param aFile
+   *        The file to read. May not be <code>null</code>.
+   * @return <code>null</code> 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;
+    }
+  }
+}

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
   {