Fix for issue #88 cxf tmp files not deleted

pom.xml

@@ -131,11 +131,6 @@
             <artifactId>peppol-sbdh</artifactId>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>no.difi.vefa</groupId>
-            <artifactId>peppol-security</artifactId>
-            <scope>provided</scope>
-        </dependency>
         <dependency>
             <groupId>no.difi.vefa</groupId>
             <artifactId>peppol-mode</artifactId>
@@ -302,6 +297,12 @@
             <groupId>org.apache.neethi</groupId>
             <artifactId>neethi</artifactId>
             <version>${neethi.version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>woodstox-core-asl</artifactId>
+                    <groupId>org.codehaus.woodstox</groupId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>
@@ -313,7 +314,7 @@
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
-            <version>1.18.8</version>
+            <version>1.18.12</version>
             <scope>provided</scope>
         </dependency>
 

src/main/java/no/difi/oxalis/as4/inbound/As4EndpointsPublisherImpl.java

@@ -1,14 +1,15 @@
 package no.difi.oxalis.as4.inbound;
 
 import com.google.inject.Inject;
+import org.apache.cxf.attachment.As4AttachmentInInterceptor;
 import org.apache.cxf.Bus;
 import org.apache.cxf.binding.soap.SoapMessage;
 import org.apache.cxf.binding.soap.SoapVersion;
 import org.apache.cxf.binding.soap.interceptor.CheckFaultInterceptor;
 import org.apache.cxf.binding.soap.interceptor.ReadHeadersInterceptor;
 import org.apache.cxf.binding.soap.interceptor.StartBodyInterceptor;
 import org.apache.cxf.ext.logging.LoggingFeature;
-import org.apache.cxf.interceptor.AttachmentInInterceptor;
+import org.apache.cxf.interceptor.StaxInEndingInterceptor;
 import org.apache.cxf.interceptor.StaxInInterceptor;
 import org.apache.cxf.jaxws.EndpointImpl;
 import org.apache.cxf.jaxws.handler.soap.SOAPHandlerFaultInInterceptor;
@@ -20,6 +21,8 @@
 import javax.xml.ws.Endpoint;
 import java.util.Arrays;
 
+import static org.apache.cxf.ws.security.SecurityConstants.ENABLE_STREAMING_SECURITY;
+
 public class As4EndpointsPublisherImpl implements As4EndpointsPublisher {
 
     @Inject
@@ -47,12 +50,15 @@ public EndpointImpl publish(Bus bus) {
                 new WSPolicyFeature());
 
         endpoint.getServer().getEndpoint().put("allow-multiplex-endpoint", Boolean.TRUE);
+        endpoint.getServer().getEndpoint().put(ENABLE_STREAMING_SECURITY, false);
         endpoint.getServer().getEndpoint()
                 .put(As4EndpointSelector.ENDPOINT_NAME, As4EndpointSelector.OXALIS_AS4_ENDPOINT_NAME);
 
         endpoint.getBinding().setHandlerChain(Arrays.asList(as4FaultInHandler, new MessagingHandler()));
         endpoint.getInInterceptors().add(oxalisAs4Interceptor);
         endpoint.getInInterceptors().add(setPolicyInInterceptor);
+        endpoint.getInInterceptors().add(new AttachmentCleanupInterceptor());
+
         endpoint.getOutInterceptors().add(setPolicyOutInterceptor);
         endpoint.getInFaultInterceptors().add(setPolicyInInterceptor);
         endpoint.getOutFaultInterceptors().add(setPolicyOutInterceptor);
@@ -64,13 +70,15 @@ protected Message createMessage(Message message) {
             }
         };
 
-        newMO.getBindingInterceptors().add(new AttachmentInInterceptor());
+        newMO.getBindingInterceptors().add(new As4AttachmentInInterceptor());
         newMO.getBindingInterceptors().add(new StaxInInterceptor());
+        newMO.getBindingInterceptors().add(new StaxInEndingInterceptor());
         newMO.getBindingInterceptors().add(new SOAPHandlerFaultInInterceptor(endpoint.getBinding()));
         newMO.getBindingInterceptors().add(new ReadHeadersInterceptor(bus, (SoapVersion) null));
         newMO.getBindingInterceptors().add(new StartBodyInterceptor());
         newMO.getBindingInterceptors().add(new CheckFaultInterceptor());
 
+
         // Add in a default selection interceptor
         newMO.getRoutingInterceptors().add(endpointSelector);
         newMO.getEndpoints().add(endpoint.getServer().getEndpoint());

src/main/java/no/difi/oxalis/as4/inbound/As4InboundHandler.java

@@ -35,6 +35,7 @@
 
 import javax.xml.soap.*;
 import javax.xml.ws.handler.MessageContext;
+import java.io.BufferedInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -334,7 +335,7 @@ private LinkedHashMap<InputStream, As4PayloadHeader> parseAttachments(Iterator<A
 
                 if (isAttachmentCompressed(partInfoHeaders, mimeHeaders)) {
                     try {
-                        is = new GZIPInputStream(is);
+                        is = new GZIPInputStream(new BufferedInputStream(is), 8192);
                     } catch (IOException e) {
                         log.info("PartInfo headers: {}", partInfoHeaders.values().stream()
                                 .map(p -> p.getName() + "=" + p.getValue())
@@ -353,35 +354,26 @@ private LinkedHashMap<InputStream, As4PayloadHeader> parseAttachments(Iterator<A
                     }
                 }
 
+                BufferedInputStream bis = new BufferedInputStream(is, 65536);
+
                 Header sbdh;
                 if (headerParser instanceof SbdhHeaderParser) {
-
-                    try {
-                        PeekingInputStream pis = new PeekingInputStream(is);
-
-                        try (SbdReader sbdReader = SbdReader.newInstance(pis)) {
-
-                            sbdh = sbdReader.getHeader();
-                            is = pis.newInputStream();
-
-                        }
-                    } catch (SbdhException | IOException e) {
-                        launderZipException(contentId, e);
-                        throw new OxalisAs4Exception("Could not extract SBDH from payload");
-                    }
+                    bis.mark(65536);
+                    sbdh = readHeader(contentId, bis);
+                    bis.reset();
                 } else {
                     sbdh = new Header()
                             .sender(ParticipantIdentifier.of(userMessage.getPartyInfo().getFrom().getPartyId().get(0).getValue()))
                             .receiver(ParticipantIdentifier.of(userMessage.getPartyInfo().getTo().getPartyId().get(0).getValue()))
-                            .documentType(DocumentTypeIdentifier.of(userMessage.getCollaborationInfo().getService().getValue(),  Scheme.of(userMessage.getCollaborationInfo().getService().getType())))
+                            .documentType(DocumentTypeIdentifier.of(userMessage.getCollaborationInfo().getService().getValue(), Scheme.of(userMessage.getCollaborationInfo().getService().getType())))
                             .identifier(InstanceIdentifier.of(userMessage.getCollaborationInfo().getAction()));
                 }
 
                 // Get an "unexpected eof in prolog"
                 As4PayloadHeader header = new As4PayloadHeader(sbdh, partInfoHeaders.values(), contentId, userMessage.getMessageInfo().getMessageId());
 
                 // Extract "fresh" InputStream
-                payloads.put(is, header);
+                payloads.put(bis, header);
 
             } catch (IOException e) {
                 throw new OxalisAs4Exception("Could not get attachment input stream", e);
@@ -391,6 +383,15 @@ private LinkedHashMap<InputStream, As4PayloadHeader> parseAttachments(Iterator<A
         return payloads;
     }
 
+    private Header readHeader(String contentId, InputStream is) throws OxalisAs4Exception {
+        try (SbdReader sbdReader = SbdReader.newInstance(is)) {
+            return sbdReader.getHeader();
+        } catch (SbdhException | IOException e) {
+            launderZipException(contentId, e);
+            throw new OxalisAs4Exception("Could not extract SBDH from payload");
+        }
+    }
+
     private void launderZipException(String contentId, Exception e) throws OxalisAs4Exception {
         Throwable cause = e;
         for (int i = 0; i < 10 && cause != null; i++) {
@@ -458,12 +459,13 @@ private Timestamp getTimestamp(SOAPHeader header) throws OxalisAs4Exception {
     }
 
     private Path persistPayload(InputStream inputStream, As4PayloadHeader as4PayloadHeader, TransmissionIdentifier ti) throws OxalisAs4Exception {
-        try {
+        try (InputStream is = inputStream) {
             // Persist content
-            Path payloadPath = persisterHandler.persist(ti, as4PayloadHeader, new UnclosableInputStream(inputStream));
+            Path payloadPath = persisterHandler.persist(ti, as4PayloadHeader, new UnclosableInputStream(is));
 
             // Exhaust InputStream
-            ByteStreams.exhaust(inputStream);
+            ByteStreams.exhaust(is);
+            inputStream.close();
             return payloadPath;
         } catch (IOException e) {
             launderZipException(as4PayloadHeader.getCid(), e);

src/main/java/no/difi/oxalis/as4/inbound/As4Interceptor.java

@@ -16,6 +16,7 @@
 import org.apache.cxf.phase.Phase;
 import org.apache.cxf.ws.policy.AssertionInfoMap;
 import org.apache.cxf.ws.security.wss4j.PolicyBasedWSS4JInInterceptor;
+import org.apache.cxf.ws.security.wss4j.PolicyBasedWSS4JStaxInInterceptor;
 import org.apache.neethi.Policy;
 import org.oasis_open.docs.ebxml_msg.ebms.v3_0.ns.core._200704.CollaborationInfo;
 import org.oasis_open.docs.ebxml_msg.ebms.v3_0.ns.core._200704.MessageInfo;
@@ -41,8 +42,7 @@ public class As4Interceptor extends AbstractSoapInterceptor {
     public As4Interceptor(PolicyService policyService) {
         super(Phase.PRE_PROTOCOL);
         this.policyService = policyService;
-        addBefore(OxalisAS4WsInInterceptor.class.getName());
-        addBefore(PolicyBasedWSS4JInInterceptor.class.getName());
+        addBefore(PolicyBasedWSS4JStaxInInterceptor.class.getName());
     }
 
     @Override

src/main/java/no/difi/oxalis/as4/inbound/As4Servlet.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 
 import static org.apache.cxf.rt.security.SecurityConstants.*;
+import static org.apache.cxf.ws.security.SecurityConstants.ENABLE_STREAMING_SECURITY;
 
 @Slf4j
 @Singleton

src/main/java/no/difi/oxalis/as4/inbound/AttachmentCleanupInterceptor.java

@@ -0,0 +1,40 @@
+package no.difi.oxalis.as4.inbound;
+
+import lombok.SneakyThrows;
+import org.apache.cxf.attachment.As4AttachmentDataSource;
+import org.apache.cxf.attachment.As4AttachmentDeserializer;
+import org.apache.cxf.interceptor.Fault;
+import org.apache.cxf.message.Attachment;
+import org.apache.cxf.message.Exchange;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.AbstractPhaseInterceptor;
+import org.apache.cxf.phase.Phase;
+
+import javax.activation.DataSource;
+
+public class AttachmentCleanupInterceptor extends AbstractPhaseInterceptor<Message> {
+
+    public AttachmentCleanupInterceptor() {
+        super(Phase.POST_INVOKE);
+    }
+
+    public void handleMessage(Message message) throws Fault {
+        Exchange exchange = message.getExchange();
+        cleanRequestAttachment(exchange);
+    }
+
+    private void cleanRequestAttachment(Exchange exchange) {
+        As4AttachmentDeserializer ad = exchange.getInMessage().get(As4AttachmentDeserializer.class);
+        ad.getRemoved().forEach(this::close);
+    }
+
+    @SneakyThrows
+    private void close(Attachment attachment) {
+        DataSource dataSource = attachment.getDataHandler().getDataSource();
+
+        if (dataSource instanceof As4AttachmentDataSource) {
+            As4AttachmentDataSource ads = (As4AttachmentDataSource) dataSource;
+            ads.closeAll();
+        }
+    }
+}