This is an automated email from the ASF dual-hosted git repository.
ffang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/main by this push:
new dbc1edb4e30 CXF-6955: Fix WS-RM retransmission of secured attachments
(#3609)
dbc1edb4e30 is described below
commit dbc1edb4e300fc03f0d57aec9a9814d5a5d5b86f
Author: Freeman(Yue) Fang <[email protected]>
AuthorDate: Thu Oct 22 08:55:57 2026 -0400
CXF-6955: Fix WS-RM retransmission of secured attachments (#3609)
* CXF-6955: Fix WS-RM retransmission of secured attachments
Save original attachments before WS-Security replaces them with single-use
secured streams, and restore them for each retransmission. Persist the
originals alongside the unsecured SOAP content so security is reapplied
consistently after recovery.
Hold recovered attachment caches across retries and release them when the
message is acknowledged or discarded, including cleanup after partial recovery
failures. Order StAX security after attachment setup so the serializer can
access the output stream.
Cover DOM and StAX signing and encryption, both action orders, secured
MTOM, persistent stores, repeated retries, and disk-backed recovery.
* CXF-6955: Address secured attachment review feedback
Report unchecked persisted-message decoding failures as IOExceptions so
recovery continues with valid messages and starts both queues. Verify recovery
after attachment-limit failures and cleanup of partial attachment caches.
Keep AttachmentSaveInterceptor inside the WS-RM capture interceptor and
preserve distinct null-ID attachments without duplicating captured originals.
Document the requirement for repeatable attachment streams in the release notes.
---
distribution/src/main/release/release_notes.txt | 4 +
.../apache/cxf/ws/rm/RMCaptureOutInterceptor.java | 42 ++
.../main/java/org/apache/cxf/ws/rm/RMManager.java | 5 +
.../org/apache/cxf/ws/rm/RMMessageConstants.java | 6 +
.../cxf/ws/rm/persistence/PersistenceUtils.java | 70 ++-
.../cxf/ws/rm/soap/RetransmissionQueueImpl.java | 29 +-
.../cxf/ws/rm/RMCaptureOutInterceptorTest.java | 57 +++
.../java/org/apache/cxf/ws/rm/RMManagerTest.java | 61 +++
.../ws/rm/persistence/PersistenceUtilsTest.java | 89 +++-
.../ws/security/wss4j/WSS4JStaxOutInterceptor.java | 3 +
.../ws/rm/sec/WSRMWithSecuredAttachmentsTest.java | 550 +++++++++++++++++++++
.../systest/ws/rm/sec/WSRMWithSecuredMtomTest.java | 334 +++++++++++++
12 files changed, 1234 insertions(+), 16 deletions(-)
diff --git a/distribution/src/main/release/release_notes.txt
b/distribution/src/main/release/release_notes.txt
index 41a75c6d8e0..24cb585ea65 100644
--- a/distribution/src/main/release/release_notes.txt
+++ b/distribution/src/main/release/release_notes.txt
@@ -29,6 +29,10 @@ New features include:
Important notes:
* CXF 4.2 now requires Java 17 as minimum
+* WS-RM attachments must be repeatable: each DataSource.getInputStream() call
+ must return a fresh stream positioned at the beginning. Persistence and
+ retransmission may read the original attachment more than once, including
+ when WS-Security signs or encrypts attachments.
Users are encouraged to review the migration guide at:
https://cxf.apache.org/docs/42-migration-guide.html
diff --git
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
index 62a40a0bc62..e3dbd56aeb2 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
@@ -22,9 +22,14 @@ package org.apache.cxf.ws.rm;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -42,6 +47,7 @@ import org.apache.cxf.interceptor.AttachmentOutInterceptor;
import org.apache.cxf.interceptor.Fault;
import org.apache.cxf.io.CachedOutputStream;
import org.apache.cxf.io.WriteOnCloseOutputStream;
+import org.apache.cxf.message.Attachment;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.ExchangeImpl;
import org.apache.cxf.message.FaultMode;
@@ -227,9 +233,45 @@ public class RMCaptureOutInterceptor extends
AbstractRMInterceptor<Message> {
}
message.getInterceptorChain().add(new CaptureStart());
+ message.getInterceptorChain().add(new
AttachmentSaveInterceptor(message.getAttachments()));
message.getInterceptorChain().add(new CaptureEnd());
}
+ /**
+ * Saves the attachments before they are replaced by the secured ones of
WS-Security, which happens in
+ * the POST_PROTOCOL phase, or while the message is written with the
streaming WS-Security. The
+ * attachments added by the marshalling, like the MTOM ones, are saved in
the POST_MARSHAL phase, but
+ * not the secured ones, which have the IDs of the attachments they
replace.
+ */
+ static class AttachmentSaveInterceptor extends
AbstractPhaseInterceptor<Message> {
+ private final List<Attachment> attachments;
+
+ AttachmentSaveInterceptor(Collection<Attachment> attachments) {
+ super(Phase.POST_MARSHAL);
+ this.attachments = attachments == null ? new ArrayList<>() : new
ArrayList<>(attachments);
+ }
+
+ @Override
+ public void handleMessage(Message message) throws Fault {
+ if (message.getAttachments() == null) {
+ return;
+ }
+ Set<String> ids = new HashSet<>();
+ Set<Attachment> saved = Collections.newSetFromMap(new
IdentityHashMap<>());
+ saved.addAll(attachments);
+ for (Attachment a : attachments) {
+ ids.add(a.getId());
+ }
+ for (Attachment a : message.getAttachments()) {
+ String id = a.getId();
+ if ((id == null || ids.add(id)) && saved.add(a)) {
+ attachments.add(a);
+ }
+ }
+ message.put(RMMessageConstants.SAVED_ATTACHMENTS, attachments);
+ }
+ }
+
private static class CaptureStart extends
AbstractPhaseInterceptor<Message> {
CaptureStart() {
super(Phase.PRE_PROTOCOL);
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
index f6d3afa5e36..e344ac7dfde 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
@@ -21,6 +21,7 @@ package org.apache.cxf.ws.rm;
import java.io.Closeable;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
@@ -639,6 +640,10 @@ public class RMManager {
// RMMessage content must be splitted into soap root message
// and attachments
PersistenceUtils.decodeRMContent(m, message);
+ if (message.getAttachments() != null) {
+ // each retransmission secures the attachments again, see
RetransmissionQueueImpl
+ message.put(RMMessageConstants.SAVED_ATTACHMENTS, new
ArrayList<>(message.getAttachments()));
+ }
RMContextUtils.setProtocolVariation(message, ss.getProtocol());
retransmissionQueue.addUnacknowledged(message);
} catch (IOException e) {
diff --git
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMMessageConstants.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMMessageConstants.java
index b06f5d69a91..0a5547a0a2b 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMMessageConstants.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMMessageConstants.java
@@ -42,6 +42,12 @@ public final class RMMessageConstants {
/** Saved Content-Type before AttachmentOutInterceptor replaces it with
multipart. */
public static final String SAVED_CONTENT_TYPE =
"org.apache.cxf.ws.rm.content.type";
+ /**
+ * Saved attachments before WS-Security may replace them with secured
ones, whose content can only be read
+ * once. Retransmissions secure the saved attachments again.
+ */
+ public static final String SAVED_ATTACHMENTS =
"org.apache.cxf.ws.rm.attachments";
+
/** Variable holds reference to source streams of the attachments.
* It must be an instance of {@link Closeable}. */
public static final String ATTACHMENTS_CLOSEABLE =
"org.apache.cxf.ws.rm.attachment.closeable";
diff --git
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/PersistenceUtils.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/PersistenceUtils.java
index dc7b137c383..3bfab6969ea 100644
---
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/PersistenceUtils.java
+++
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/PersistenceUtils.java
@@ -23,6 +23,8 @@ import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
@@ -30,11 +32,13 @@ import java.util.logging.Logger;
import javax.xml.stream.XMLStreamReader;
+import jakarta.activation.DataSource;
import jakarta.activation.MimeType;
import jakarta.activation.MimeTypeParseException;
import jakarta.xml.bind.JAXBContext;
import jakarta.xml.bind.JAXBElement;
import jakarta.xml.bind.JAXBException;
+import org.apache.cxf.attachment.AttachmentDataSource;
import org.apache.cxf.attachment.AttachmentDeserializer;
import org.apache.cxf.attachment.AttachmentSerializer;
import org.apache.cxf.attachment.AttachmentUtil;
@@ -44,6 +48,7 @@ import org.apache.cxf.helpers.CastUtils;
import org.apache.cxf.helpers.IOUtils;
import org.apache.cxf.helpers.LoadingByteArrayOutputStream;
import org.apache.cxf.io.CachedOutputStream;
+import org.apache.cxf.message.Attachment;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageImpl;
import org.apache.cxf.message.MessageUtils;
@@ -225,7 +230,8 @@ public final class PersistenceUtils {
*/
public static void encodeRMContent(RMMessage rmmsg, Message msg,
CachedOutputStream msgContent)
throws IOException {
- if (msg.getAttachments() == null || msg.getAttachments().isEmpty()) {
+ Collection<Attachment> attachments = getAttachments(msg);
+ if (attachments == null || attachments.isEmpty()) {
rmmsg.setContentType(getRootContentType(msg));
rmmsg.setContent(msgContent);
} else {
@@ -233,10 +239,20 @@ public final class PersistenceUtils {
}
}
+ /**
+ * Attachments of the message to persist: for a captured outbound message
the saved ones, as WS-Security may
+ * have replaced the attachments of the message with secured ones, and the
SOAP content is persisted unsecured.
+ */
+ private static Collection<Attachment> getAttachments(Message msg) {
+ Collection<Attachment> saved =
CastUtils.cast((Collection<?>)msg.get(RMMessageConstants.SAVED_ATTACHMENTS));
+ return saved != null ? saved : msg.getAttachments();
+ }
+
public static void encodeRMContent(RMMessage rmmsg, Message msg,
InputStream msgContent)
throws IOException {
CachedOutputStream cos = new CachedOutputStream();
- if (msg.getAttachments() == null || msg.getAttachments().isEmpty()) {
+ Collection<Attachment> attachments = getAttachments(msg);
+ if (attachments == null || attachments.isEmpty()) {
rmmsg.setContentType(getRootContentType(msg));
IOUtils.copyAndCloseInput(msgContent, cos);
cos.flush();
@@ -244,7 +260,7 @@ public final class PersistenceUtils {
} else {
MessageImpl msgImpl1 = new MessageImpl();
msgImpl1.setContent(OutputStream.class, cos);
- msgImpl1.setAttachments(msg.getAttachments());
+ msgImpl1.setAttachments(attachments);
// an outbound MTOM message is stored as an XOP package, so that
it is resent as one
boolean xop = MessageUtils.isOutbound(msg) &&
AttachmentUtil.isMtomEnabled(msg);
String rootType = getRootContentType(msg);
@@ -270,6 +286,16 @@ public final class PersistenceUtils {
}
public static void decodeRMContent(RMMessage rmmsg, Message msg) throws
IOException {
+ try {
+ doDecodeRMContent(rmmsg, msg);
+ } catch (RuntimeException e) {
+ // Lazy attachment parsing and cache limits can fail with
unchecked exceptions.
+ // Recovery handles a bad persisted message as an I/O failure and
continues with the rest.
+ throw new IOException("Unable to decode persisted RM message", e);
+ }
+ }
+
+ private static void doDecodeRMContent(RMMessage rmmsg, Message msg) throws
IOException {
String contentType = rmmsg.getContentType();
final CachedOutputStream cos = rmmsg.getContent();
if ((null != contentType) &&
contentType.startsWith("multipart/related")) {
@@ -283,7 +309,6 @@ public final class PersistenceUtils {
IOUtils.copy(msg.getContent(InputStream.class), cosSoap);
cosSoap.flush();
msg.put(RMMessageConstants.SAVED_CONTENT, cosSoap);
-
// restore the SOAP root content type, the multipart type is
recreated when the message is sent
String rootType = getRootContentType(msg);
String xopWrappedType = getXopWrappedType(rootType);
@@ -295,12 +320,22 @@ public final class PersistenceUtils {
msg.put(RMMessageConstants.SAVED_CONTENT_TYPE, rootType);
msg.put(Message.CONTENT_TYPE, rootType);
- // REVISIT -- At the moment references must be hold for
retransmission
- // and the final cleanup of the CachedOutputStream.
- msg.put(RMMessageConstants.ATTACHMENTS_CLOSEABLE, new Closeable() {
-
+ List<AttachmentDataSource> attachmentCaches = new ArrayList<>();
+ Closeable cleanup = new Closeable() {
@Override
public void close() throws IOException {
+ for (AttachmentDataSource ads : attachmentCaches) {
+ ads.release();
+ // Closing a reader after releasing the hold deletes
an otherwise unused cache file.
+ InputStream cached = ads.getInputStream();
+ if (cached != null) {
+ try {
+ cached.close();
+ } catch (IOException e) {
+ // Ignore, but continue releasing the other
caches and the persisted message.
+ }
+ }
+ }
try {
is.close();
} catch (IOException e) {
@@ -313,7 +348,24 @@ public final class PersistenceUtils {
}
}
- });
+ };
+ msg.put(RMMessageConstants.ATTACHMENTS_CLOSEABLE, cleanup);
+ // Keep disk-backed attachment caches until the message is
acknowledged or discarded.
+ // Caching alone allows the last reader to delete the file after
the first retransmission.
+ try {
+ for (Attachment a : msg.getAttachments()) {
+ DataSource ds = a.getDataHandler().getDataSource();
+ if (ds instanceof AttachmentDataSource) {
+ AttachmentDataSource ads = (AttachmentDataSource)ds;
+ ads.hold(msg);
+ attachmentCaches.add(ads);
+ }
+ }
+ } catch (IOException | RuntimeException e) {
+ cleanup.close();
+ cosSoap.close();
+ throw e;
+ }
} else {
msg.put(RMMessageConstants.SAVED_CONTENT, cos);
msg.put(Message.CONTENT_TYPE, contentType);
diff --git
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
index 9fb8dc35b38..7db6a9dcbf6 100644
---
a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
+++
b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
@@ -57,6 +57,7 @@ import
org.apache.cxf.binding.soap.interceptor.SoapOutInterceptor;
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.endpoint.DeferredConduitSelector;
import org.apache.cxf.endpoint.Endpoint;
+import org.apache.cxf.helpers.CastUtils;
import org.apache.cxf.helpers.DOMUtils;
import org.apache.cxf.interceptor.AbstractOutDatabindingInterceptor;
import org.apache.cxf.interceptor.AttachmentOutInterceptor;
@@ -65,6 +66,7 @@ import org.apache.cxf.interceptor.Interceptor;
import org.apache.cxf.io.CachedOutputStream;
import org.apache.cxf.io.CachedOutputStreamCallback;
import org.apache.cxf.io.WriteOnCloseOutputStream;
+import org.apache.cxf.message.Attachment;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageUtils;
import org.apache.cxf.phase.Phase;
@@ -791,6 +793,25 @@ public class RetransmissionQueueImpl implements
RetransmissionQueue {
}
}
+ /**
+ * Prepares the message for its attachments to be written again. The flags
of the attachment out interceptor
+ * are cleared and the Content-Type it replaced is restored. The
attachments saved before WS-Security replaced
+ * them with secured ones, whose content can only be read once, are
restored, as the retransmission secures
+ * them again.
+ */
+ private static void resetAttachments(SoapMessage message) {
+ message.remove(AttachmentOutInterceptor.ATTACHMENT_OUT_CHECKED);
+ String savedContentType =
(String)message.get(RMMessageConstants.SAVED_CONTENT_TYPE);
+ if (savedContentType != null) {
+ message.put(org.apache.cxf.message.Message.CONTENT_TYPE,
savedContentType);
+ }
+ Collection<Attachment> saved =
+
CastUtils.cast((Collection<?>)message.get(RMMessageConstants.SAVED_ATTACHMENTS));
+ if (saved != null) {
+ message.setAttachments(new ArrayList<>(saved));
+ }
+ }
+
private void doResend(SoapMessage message) {
InputStream is = null;
try {
@@ -811,13 +832,9 @@ public class RetransmissionQueueImpl implements
RetransmissionQueue {
}
message.setInterceptorChain(retransmitChain);
- // clear flags for SOAP and attachment out interceptors so
envelope and attachments will be written
+ // clear flag for SOAP out interceptor so envelope will be written
message.remove(SoapOutInterceptor.WROTE_ENVELOPE_START);
- message.remove(AttachmentOutInterceptor.ATTACHMENT_OUT_CHECKED);
- String savedContentType =
(String)message.get(RMMessageConstants.SAVED_CONTENT_TYPE);
- if (savedContentType != null) {
- message.put(org.apache.cxf.message.Message.CONTENT_TYPE,
savedContentType);
- }
+ resetAttachments(message);
// discard all saved content
Set<Class<?>> formats = message.getContentFormats();
diff --git
a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptorTest.java
b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptorTest.java
new file mode 100644
index 00000000000..7cf451c1a56
--- /dev/null
+++
b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptorTest.java
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.cxf.ws.rm;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.cxf.attachment.AttachmentImpl;
+import org.apache.cxf.message.Attachment;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.message.MessageImpl;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+public class RMCaptureOutInterceptorTest {
+ @Test
+ public void testSaveAttachmentsWithNullIds() {
+ Attachment original = new AttachmentImpl(null);
+ Attachment identified = new AttachmentImpl("attachment-1");
+ List<Attachment> originals = new ArrayList<>(List.of(original,
identified));
+ RMCaptureOutInterceptor.AttachmentSaveInterceptor interceptor =
+ new RMCaptureOutInterceptor.AttachmentSaveInterceptor(originals);
+
+ Attachment first = new AttachmentImpl(null);
+ Attachment second = new AttachmentImpl(null);
+ Message message = new MessageImpl();
+ message.setAttachments(List.of(original, new
AttachmentImpl("attachment-1"), first, second));
+ interceptor.handleMessage(message);
+
+ Object saved = message.get(RMMessageConstants.SAVED_ATTACHMENTS);
+ assertEquals(List.of(original, identified, first, second), saved);
+ assertEquals(2, originals.size());
+ interceptor.handleMessage(message);
+ assertSame(saved, message.get(RMMessageConstants.SAVED_ATTACHMENTS));
+ assertEquals(List.of(original, identified, first, second), saved);
+ }
+}
diff --git a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java
b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java
index 4579a30eea6..e456f76963d 100644
--- a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java
+++ b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java
@@ -28,12 +28,17 @@ import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.TimerTask;
import javax.xml.namespace.QName;
+import jakarta.activation.DataHandler;
+import jakarta.mail.util.ByteArrayDataSource;
import org.apache.cxf.Bus;
+import org.apache.cxf.attachment.AttachmentDeserializer;
+import org.apache.cxf.attachment.AttachmentImpl;
import org.apache.cxf.binding.Binding;
import org.apache.cxf.binding.soap.SoapBinding;
import org.apache.cxf.bus.spring.SpringBusFactory;
@@ -43,8 +48,10 @@ import org.apache.cxf.endpoint.Server;
import org.apache.cxf.helpers.CastUtils;
import org.apache.cxf.helpers.IOUtils;
import org.apache.cxf.io.CachedOutputStream;
+import org.apache.cxf.message.Attachment;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.Message;
+import org.apache.cxf.message.MessageImpl;
import org.apache.cxf.service.Service;
import org.apache.cxf.service.model.BindingInfo;
import org.apache.cxf.service.model.EndpointInfo;
@@ -59,6 +66,7 @@ import org.apache.cxf.ws.addressing.RelatesToType;
import org.apache.cxf.ws.rm.manager.DestinationPolicyType;
import org.apache.cxf.ws.rm.manager.SequenceTerminationPolicyType;
import org.apache.cxf.ws.rm.manager.SourcePolicyType;
+import org.apache.cxf.ws.rm.persistence.PersistenceUtils;
import org.apache.cxf.ws.rm.persistence.RMMessage;
import org.apache.cxf.ws.rm.persistence.RMStore;
import org.apache.cxf.ws.rm.v200702.CreateSequenceResponseType;
@@ -518,6 +526,59 @@ public class RMManagerTest {
assertEquals(1, msg.getAttachments().size());
}
+ @Test
+ public void testRecoveryContinuesAfterInvalidAttachments() throws
IOException {
+ manager = spy(new RMManager());
+ manager.setReliableEndpointsMap(new HashMap<>());
+ Endpoint endpoint = mock(Endpoint.class);
+ setUpEndpointForRecovery(endpoint, mock(EndpointInfo.class),
mock(ServiceInfo.class),
+ mock(BindingInfo.class), mock(InterfaceInfo.class));
+ Conduit conduit = mock(Conduit.class);
+ SourceSequence source = mock(SourceSequence.class);
+ DestinationSequence destination = mock(DestinationSequence.class);
+ RMMessage validSource = createRecoveryMessage(1, 2L);
+ setUpRecoverReliableEndpointWithAttachment(endpoint, conduit, source,
destination,
+ validSource, ArgumentCaptor.forClass(Message.class));
+ int tooMany = AttachmentDeserializer.DEFAULT_ATTACHMENT_MAX_COUNT + 2;
+ when(manager.getStore().getMessages(source.getIdentifier(), true))
+ .thenReturn(List.of(createRecoveryMessage(tooMany, 1L),
validSource));
+ Identifier destinationId = new Identifier();
+ destinationId.setValue("D1");
+ when(destination.getIdentifier()).thenReturn(destinationId);
+ when(manager.getStore().getMessages(destinationId, false))
+ .thenReturn(List.of(createRecoveryMessage(tooMany, 1L),
createRecoveryMessage(1, 2L)));
+
+ manager.recoverReliableEndpoint(endpoint, conduit);
+
+ ArgumentCaptor<Message> outgoing =
ArgumentCaptor.forClass(Message.class);
+
verify(manager.getRetransmissionQueue()).addUnacknowledged(outgoing.capture());
+ assertEquals(Long.valueOf(2L),
RMContextUtils.retrieveRMProperties(outgoing.getValue(), true)
+ .getSequence().getMessageNumber());
+ ArgumentCaptor<Message> incoming =
ArgumentCaptor.forClass(Message.class);
+
verify(manager.getRedeliveryQueue()).addUndelivered(incoming.capture());
+ assertEquals(Long.valueOf(2L),
RMContextUtils.retrieveRMProperties(incoming.getValue(), false)
+ .getSequence().getMessageNumber());
+ verify(destination).addDeliveringMessageNumber(2L);
+ verify(manager.getRetransmissionQueue()).start();
+ verify(manager.getRedeliveryQueue()).start();
+ }
+
+ private static RMMessage createRecoveryMessage(int count, long number)
throws IOException {
+ Message message = new MessageImpl();
+ message.put(Message.CONTENT_TYPE, "text/xml");
+ List<Attachment> attachments = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ attachments.add(new AttachmentImpl("attachment-" + i,
+ new DataHandler(new ByteArrayDataSource("attachment content",
"text/plain"))));
+ }
+ message.setAttachments(attachments);
+ RMMessage persisted = new RMMessage();
+ persisted.setMessageNumber(number);
+ PersistenceUtils.encodeRMContent(persisted, message,
+ new
ByteArrayInputStream("<root/>".getBytes(StandardCharsets.UTF_8)));
+ return persisted;
+ }
+
Runnable setUpRecoverReliableEndpointWithAttachment(Endpoint endpoint,
Conduit conduit,
SourceSequence ss,
diff --git
a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/PersistenceUtilsTest.java
b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/PersistenceUtilsTest.java
index c765c626bcb..915c2d331ec 100644
---
a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/PersistenceUtilsTest.java
+++
b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/PersistenceUtilsTest.java
@@ -20,6 +20,8 @@
package org.apache.cxf.ws.rm.persistence;
import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -36,6 +38,7 @@ import org.apache.cxf.attachment.AttachmentDeserializer;
import org.apache.cxf.attachment.AttachmentImpl;
import org.apache.cxf.attachment.AttachmentSerializer;
import org.apache.cxf.helpers.IOUtils;
+import org.apache.cxf.io.CacheSizeExceededException;
import org.apache.cxf.io.CachedOutputStream;
import org.apache.cxf.message.Attachment;
import org.apache.cxf.message.Exchange;
@@ -46,12 +49,15 @@ import org.apache.cxf.ws.rm.RMMessageConstants;
import org.apache.cxf.ws.rm.v200702.SequenceAcknowledgement;
import
org.apache.cxf.ws.rm.v200702.SequenceAcknowledgement.AcknowledgementRange;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
/**
@@ -66,6 +72,9 @@ public class PersistenceUtilsTest {
private static final String SOAP_PART = "<soap:Envelope
xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\";>"
+ "<data/></soap:Envelope>";
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
@Test
public void testSerialiseDeserialiseAcknowledgement() {
SequenceAcknowledgement ack = new SequenceAcknowledgement();
@@ -191,6 +200,84 @@ public class PersistenceUtilsTest {
restored.startsWith("multipart/"));
}
+ @Test
+ public void testRecoveredAttachmentOnDiskCanBeReadRepeatedly() throws
Exception {
+ checkRecoveredAttachmentOnDisk(3);
+ }
+
+ @Test
+ public void testRecoveredAttachmentOnDiskDiscardedWithoutReading() throws
Exception {
+ checkRecoveredAttachmentOnDisk(0);
+ }
+
+ @Test
+ public void testRecoveredAttachmentCachesReleasedOnFailure() throws
Exception {
+ Message source = new MessageImpl();
+ source.put(Message.CONTENT_TYPE, "text/xml");
+ addAttachment(source);
+ source.getAttachments().add(new AttachmentImpl("large.txt",
+ new DataHandler(new ByteArrayDataSource("too large for the
configured attachment limit", "text/plain"))));
+ RMMessage persisted = new RMMessage();
+ PersistenceUtils.encodeRMContent(persisted, source,
+ new
ByteArrayInputStream(SOAP_PART.getBytes(StandardCharsets.UTF_8)));
+
+ File directory = temporaryFolder.newFolder();
+ Message restored = new MessageImpl();
+ restored.put(AttachmentDeserializer.ATTACHMENT_MEMORY_THRESHOLD, 1);
+ restored.put(AttachmentDeserializer.ATTACHMENT_DIRECTORY, directory);
+ restored.put(AttachmentDeserializer.ATTACHMENT_MAX_SIZE, 16L);
+ IOException error = assertThrows(IOException.class,
+ () -> PersistenceUtils.decodeRMContent(persisted, restored));
+ assertTrue(error.getCause() instanceof CacheSizeExceededException);
+ assertEquals("partial recovery must release the attachment caches", 0,
directory.list().length);
+ }
+
+ @Test
+ public void testTooManyRecoveredAttachmentsReportedAsIOException() throws
Exception {
+ Message source = new MessageImpl();
+ source.put(Message.CONTENT_TYPE, "text/xml");
+ Collection<Attachment> attachments = new ArrayList<>();
+ for (int i = 0; i <
AttachmentDeserializer.DEFAULT_ATTACHMENT_MAX_COUNT + 2; i++) {
+ attachments.add(new AttachmentImpl("attachment-" + i,
+ new DataHandler(new ByteArrayDataSource("attachment content",
"text/plain"))));
+ }
+ source.setAttachments(attachments);
+ RMMessage persisted = new RMMessage();
+ PersistenceUtils.encodeRMContent(persisted, source,
+ new
ByteArrayInputStream(SOAP_PART.getBytes(StandardCharsets.UTF_8)));
+
+ File directory = temporaryFolder.newFolder();
+ Message restored = new MessageImpl();
+ restored.put(AttachmentDeserializer.ATTACHMENT_MEMORY_THRESHOLD, 1);
+ restored.put(AttachmentDeserializer.ATTACHMENT_DIRECTORY, directory);
+ IOException error = assertThrows(IOException.class,
+ () -> PersistenceUtils.decodeRMContent(persisted, restored));
+ assertTrue(error.getCause() instanceof RuntimeException);
+ assertTrue(error.getCause().getCause() instanceof IOException);
+ assertEquals("partial recovery must release the attachment caches", 0,
directory.list().length);
+ }
+
+ private void checkRecoveredAttachmentOnDisk(int reads) throws Exception {
+ File directory = temporaryFolder.newFolder();
+ Message restored = new MessageImpl();
+ restored.put(AttachmentDeserializer.ATTACHMENT_MEMORY_THRESHOLD, 1);
+ restored.put(AttachmentDeserializer.ATTACHMENT_DIRECTORY, directory);
+ PersistenceUtils.decodeRMContent(encodeOutbound("text/xml", false),
restored);
+ Closeable cleanup =
(Closeable)restored.get(RMMessageConstants.ATTACHMENTS_CLOSEABLE);
+ try {
+ assertTrue("the attachment must spill to disk",
directory.list().length > 0);
+ Attachment attachment =
restored.getAttachments().iterator().next();
+ for (int i = 0; i < reads; i++) {
+ try (InputStream in =
attachment.getDataHandler().getInputStream()) {
+ assertEquals("hello world!", IOUtils.toString(in));
+ }
+ }
+ } finally {
+ cleanup.close();
+ }
+ assertEquals("attachment cache files must be deleted on cleanup", 0,
directory.list().length);
+ }
+
@Test
public void testEncodeDecodeInboundPathViaPartHeaders() throws Exception {
// Simulates the server (destination) side: inbound message has no
SAVED_CONTENT_TYPE
@@ -502,4 +589,4 @@ public class PersistenceUtilsTest {
}
return false;
}
-}
\ No newline at end of file
+}
diff --git
a/rt/ws/security/src/main/java/org/apache/cxf/ws/security/wss4j/WSS4JStaxOutInterceptor.java
b/rt/ws/security/src/main/java/org/apache/cxf/ws/security/wss4j/WSS4JStaxOutInterceptor.java
index e3f5f5ae147..36e3b16824a 100644
---
a/rt/ws/security/src/main/java/org/apache/cxf/ws/security/wss4j/WSS4JStaxOutInterceptor.java
+++
b/rt/ws/security/src/main/java/org/apache/cxf/ws/security/wss4j/WSS4JStaxOutInterceptor.java
@@ -75,6 +75,7 @@ public class WSS4JStaxOutInterceptor extends
AbstractWSS4JStaxInterceptor {
setPhase(Phase.PRE_STREAM);
getBefore().add(StaxOutInterceptor.class.getName());
getAfter().add("org.apache.cxf.ext.logging.LoggingOutInterceptor");
+ getAfter().add(AttachmentOutInterceptor.class.getName());
ending = createEndingInterceptor();
}
@@ -84,6 +85,7 @@ public class WSS4JStaxOutInterceptor extends
AbstractWSS4JStaxInterceptor {
setPhase(Phase.PRE_STREAM);
getBefore().add(StaxOutInterceptor.class.getName());
getAfter().add("org.apache.cxf.ext.logging.LoggingOutInterceptor");
+ getAfter().add(AttachmentOutInterceptor.class.getName());
ending = createEndingInterceptor();
}
@@ -93,6 +95,7 @@ public class WSS4JStaxOutInterceptor extends
AbstractWSS4JStaxInterceptor {
setPhase(Phase.PRE_STREAM);
getBefore().add(StaxOutInterceptor.class.getName());
getAfter().add("org.apache.cxf.ext.logging.LoggingOutInterceptor");
+ getAfter().add(AttachmentOutInterceptor.class.getName());
ending = createEndingInterceptor();
}
diff --git
a/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/sec/WSRMWithSecuredAttachmentsTest.java
b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/sec/WSRMWithSecuredAttachmentsTest.java
new file mode 100644
index 00000000000..c5206843f70
--- /dev/null
+++
b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/sec/WSRMWithSecuredAttachmentsTest.java
@@ -0,0 +1,550 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.cxf.systest.ws.rm.sec;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.activation.DataHandler;
+import jakarta.mail.util.ByteArrayDataSource;
+import jakarta.xml.ws.Endpoint;
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
+import org.apache.cxf.attachment.AttachmentDeserializer;
+import org.apache.cxf.attachment.AttachmentImpl;
+import org.apache.cxf.binding.soap.SoapMessage;
+import org.apache.cxf.bus.spring.SpringBusFactory;
+import org.apache.cxf.endpoint.Client;
+import org.apache.cxf.frontend.ClientProxy;
+import org.apache.cxf.greeter_control.Greeter;
+import org.apache.cxf.greeter_control.GreeterService;
+import org.apache.cxf.interceptor.Fault;
+import org.apache.cxf.interceptor.Interceptor;
+import org.apache.cxf.jaxws.EndpointImpl;
+import org.apache.cxf.message.Attachment;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.AbstractPhaseInterceptor;
+import org.apache.cxf.phase.Phase;
+import org.apache.cxf.service.model.BindingOperationInfo;
+import org.apache.cxf.systest.ws.rm.GreeterImpl;
+import org.apache.cxf.systest.ws.rm.MessageLossSimulator;
+import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
+import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
+import org.apache.cxf.testutil.common.TestUtil;
+import org.apache.cxf.ws.rm.RMManager;
+import org.apache.cxf.ws.security.wss4j.WSS4JInInterceptor;
+import org.apache.cxf.ws.security.wss4j.WSS4JOutInterceptor;
+import org.apache.cxf.ws.security.wss4j.WSS4JStaxOutInterceptor;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests the WS-RM retransmission of messages whose attachments are encrypted
or signed with WS-Security.
+ * The security processing replaces the attachments with secured ones whose
content can only be read once,
+ * so each retransmission must secure the original attachments again. Each
test is run with the DOM and
+ * the StAX WS-Security interceptors on the client.
+ */
+public class WSRMWithSecuredAttachmentsTest extends
AbstractBusClientServerTestBase {
+ public static final String PORT =
allocatePort(WSRMWithSecuredAttachmentsTest.class);
+
+ static final byte[] ATTACHMENT = "attachment content of
CXF-6955".getBytes(StandardCharsets.UTF_8);
+ /** The content of the attachments received by the server, after the
security processing */
+ static final BlockingQueue<byte[]> RECEIVED = new LinkedBlockingQueue<>();
+ /** The number of the next requests with attachments the server rejects */
+ static final AtomicInteger REJECTIONS = new AtomicInteger();
+
+ private static final String RM_CONFIG =
"/org/apache/cxf/systest/ws/rm/simple.xml";
+ private static final String PERSISTENT_CONFIG =
"/org/apache/cxf/systest/ws/rm/persistent.xml";
+ private static final String DB_NAME = "wsrm-sec-att-client";
+ /** Each test uses its own in-memory database, as an in-memory database
lives as long as the JVM */
+ private static final AtomicInteger DB_COUNT = new AtomicInteger();
+
+ private Bus bus;
+
+ public static class Server extends AbstractBusTestServerBase {
+ protected void run() {
+ Bus bus = new SpringBusFactory().createBus(RM_CONFIG);
+ BusFactory.setDefaultBus(bus);
+ setBus(bus);
+ publish("encrypt", "Encrypt");
+ publish("sign", "Signature");
+ publish("none", null);
+ publish("sign-encrypt", "Signature Encrypt");
+ publish("encrypt-sign", "Encrypt Signature");
+ }
+
+ private static void publish(String path, String action) {
+ EndpointImpl ep = (EndpointImpl)Endpoint.create(new GreeterImpl());
+ ep.getInInterceptors().add(new Rejector());
+ if (action != null) {
+ Map<String, Object> props = new HashMap<>();
+ props.put("action", action);
+ props.put("decryptionPropFile", "bob.properties");
+ props.put("signatureVerificationPropFile", "bob.properties");
+ props.put("passwordCallbackClass",
UTPasswordCallback.class.getName());
+ ep.getInInterceptors().add(new
AttachmentsWSS4JInInterceptor(props));
+ }
+ ep.getInInterceptors().add(new AttachmentRecorder());
+ ep.publish("http://localhost:"; + PORT + "/" + path);
+ }
+ }
+
+ /**
+ * Only processes the messages with attachments, as the RM protocol
messages are not secured by the
+ * client (its security interceptors are only on the application client).
+ */
+ static class AttachmentsWSS4JInInterceptor extends WSS4JInInterceptor {
+ AttachmentsWSS4JInInterceptor(Map<String, Object> props) {
+ super(props);
+ }
+
+ @Override
+ public void handleMessage(SoapMessage message) {
+ if (isMultipart(message)) {
+ super.handleMessage(message);
+ }
+ }
+ }
+
+ /**
+ * Only secures the messages with attachments, as the StAX interceptor
fails on messages without the
+ * attachments to secure, like the RM protocol messages of a recovery,
which use the endpoint interceptors.
+ */
+ static class AttachmentsWSS4JStaxOutInterceptor extends
WSS4JStaxOutInterceptor {
+ AttachmentsWSS4JStaxOutInterceptor(Map<String, Object> props) {
+ super(props);
+ }
+
+ @Override
+ public void handleMessage(SoapMessage message) {
+ if (message.getAttachments() != null &&
!message.getAttachments().isEmpty()) {
+ super.handleMessage(message);
+ }
+ }
+ }
+
+ /** Rejects the next REJECTIONS requests with attachments, so that the
client retransmits them again */
+ static class Rejector extends AbstractPhaseInterceptor<Message> {
+ Rejector() {
+ super(Phase.RECEIVE);
+ }
+
+ public void handleMessage(Message message) {
+ if (isMultipart(message) && REJECTIONS.getAndUpdate(n ->
Math.max(0, n - 1)) > 0) {
+ throw new Fault(new IllegalStateException("simulated
failure"));
+ }
+ }
+ }
+
+ /** Records the content of the attachments received after the security
processing */
+ static class AttachmentRecorder extends AbstractPhaseInterceptor<Message> {
+ AttachmentRecorder() {
+ super(Phase.PRE_INVOKE);
+ }
+
+ public void handleMessage(Message message) {
+ if (message.getAttachments() == null) {
+ return;
+ }
+ for (Attachment a : message.getAttachments()) {
+ try (InputStream in = a.getDataHandler().getInputStream()) {
+ RECEIVED.add(in.readAllBytes());
+ } catch (Exception e) {
+ RECEIVED.add(("failed to read the attachment: " +
e).getBytes(StandardCharsets.UTF_8));
+ }
+ }
+ }
+ }
+
+ /** Adds a SwA attachment to the greetMeOneWay requests */
+ static class AttachmentAdder extends AbstractPhaseInterceptor<Message> {
+ AttachmentAdder() {
+ super(Phase.PRE_LOGICAL);
+ }
+
+ public void handleMessage(Message message) {
+ BindingOperationInfo boi =
message.getExchange().getBindingOperationInfo();
+ if (boi == null ||
!"greetMeOneWay".equals(boi.getName().getLocalPart())) {
+ return;
+ }
+ List<Attachment> attachments = new ArrayList<>();
+ attachments.add(new AttachmentImpl("[email protected]",
+ new DataHandler(new ByteArrayDataSource(ATTACHMENT,
"text/plain"))));
+ message.setAttachments(attachments);
+ }
+ }
+
+ static boolean isMultipart(Message message) {
+ String contentType = (String)message.get(Message.CONTENT_TYPE);
+ return contentType != null &&
contentType.startsWith("multipart/related");
+ }
+
+ @BeforeClass
+ public static void startServers() throws Exception {
+ assertTrue("server did not launch correctly",
launchServer(Server.class, true));
+ }
+
+ @Before
+ public void clearReceived() {
+ RECEIVED.clear();
+ REJECTIONS.set(0);
+ }
+
+ @After
+ public void shutdownBus() {
+ REJECTIONS.set(0);
+ if (bus != null) {
+ bus.shutdown(true);
+ bus = null;
+ }
+ }
+
+ @Test
+ public void testPlainAttachmentRetransmitted() throws Exception {
+ sendAndAssertReceived("none", null, false, true);
+ }
+
+ @Test
+ public void testEncryptedAttachmentWithoutLoss() throws Exception {
+ sendAndAssertReceived("encrypt", "Encrypt", false, false);
+ }
+
+ @Test
+ public void testEncryptedAttachmentRetransmitted() throws Exception {
+ sendAndAssertReceived("encrypt", "Encrypt", false, true);
+ }
+
+ @Test
+ public void testEncryptedAttachmentRetransmittedTwice() throws Exception {
+ REJECTIONS.set(1);
+ sendAndAssertReceived("encrypt", "Encrypt", false, true);
+ }
+
+ @Test
+ public void testSignedAttachmentWithoutLoss() throws Exception {
+ sendAndAssertReceived("sign", "Signature", false, false);
+ }
+
+ @Test
+ public void testSignedAttachmentRetransmitted() throws Exception {
+ sendAndAssertReceived("sign", "Signature", false, true);
+ }
+
+ @Test
+ public void testSignedAttachmentRetransmittedTwice() throws Exception {
+ REJECTIONS.set(1);
+ sendAndAssertReceived("sign", "Signature", false, true);
+ }
+
+ /** The message is persisted with its attachment while it is sent */
+ @Test
+ public void testEncryptedAttachmentWithPersistentStore() throws Exception {
+ createPersistentBus(DB_NAME + "-" + DB_COUNT.getAndIncrement(), 2000L,
false);
+ sendAndAssertReceived("encrypt", "Encrypt", false);
+ }
+
+ @Test
+ public void testSignedAttachmentWithPersistentStore() throws Exception {
+ createPersistentBus(DB_NAME + "-" + DB_COUNT.getAndIncrement(), 2000L,
false);
+ sendAndAssertReceived("sign", "Signature", false);
+ }
+
+ @Test
+ public void testEncryptedAttachmentRetransmittedAfterRecovery() throws
Exception {
+ sendRecoverAndAssertReceived("encrypt", "Encrypt", false);
+ }
+
+ @Test
+ public void testSignedAttachmentRetransmittedAfterRecovery() throws
Exception {
+ sendRecoverAndAssertReceived("sign", "Signature", false);
+ }
+
+ /** The first retransmission of the recovered message is rejected, so it
is retransmitted again */
+ @Test
+ public void testEncryptedAttachmentRetransmittedTwiceAfterRecovery()
throws Exception {
+ REJECTIONS.set(1);
+ sendRecoverAndAssertReceived("encrypt", "Encrypt", false);
+ }
+
+ @Test
+ public void testSignedAttachmentRetransmittedTwiceAfterRecovery() throws
Exception {
+ REJECTIONS.set(1);
+ sendRecoverAndAssertReceived("sign", "Signature", false);
+ }
+
+ @Test
+ public void testStaxEncryptedAttachmentWithoutLoss() throws Exception {
+ sendAndAssertReceived("encrypt", "Encrypt", true, false);
+ }
+
+ @Test
+ public void testStaxEncryptedAttachmentRetransmitted() throws Exception {
+ sendAndAssertReceived("encrypt", "Encrypt", true, true);
+ }
+
+ @Test
+ public void testStaxEncryptedAttachmentRetransmittedTwice() throws
Exception {
+ REJECTIONS.set(1);
+ sendAndAssertReceived("encrypt", "Encrypt", true, true);
+ }
+
+ @Test
+ public void testStaxSignedAttachmentWithoutLoss() throws Exception {
+ sendAndAssertReceived("sign", "Signature", true, false);
+ }
+
+ @Test
+ public void testStaxSignedAttachmentRetransmitted() throws Exception {
+ sendAndAssertReceived("sign", "Signature", true, true);
+ }
+
+ @Test
+ public void testStaxSignedAttachmentRetransmittedTwice() throws Exception {
+ REJECTIONS.set(1);
+ sendAndAssertReceived("sign", "Signature", true, true);
+ }
+
+ @Test
+ public void testStaxEncryptedAttachmentWithPersistentStore() throws
Exception {
+ createPersistentBus(DB_NAME + "-" + DB_COUNT.getAndIncrement(), 2000L,
false);
+ sendAndAssertReceived("encrypt", "Encrypt", true);
+ }
+
+ @Test
+ public void testStaxSignedAttachmentWithPersistentStore() throws Exception
{
+ createPersistentBus(DB_NAME + "-" + DB_COUNT.getAndIncrement(), 2000L,
false);
+ sendAndAssertReceived("sign", "Signature", true);
+ }
+
+ @Test
+ public void testStaxEncryptedAttachmentRetransmittedAfterRecovery() throws
Exception {
+ sendRecoverAndAssertReceived("encrypt", "Encrypt", true);
+ }
+
+ @Test
+ public void testStaxSignedAttachmentRetransmittedAfterRecovery() throws
Exception {
+ sendRecoverAndAssertReceived("sign", "Signature", true);
+ }
+
+ @Test
+ public void testStaxEncryptedAttachmentRetransmittedTwiceAfterRecovery()
throws Exception {
+ REJECTIONS.set(1);
+ sendRecoverAndAssertReceived("encrypt", "Encrypt", true);
+ }
+
+ @Test
+ public void testStaxSignedAttachmentRetransmittedTwiceAfterRecovery()
throws Exception {
+ REJECTIONS.set(1);
+ sendRecoverAndAssertReceived("sign", "Signature", true);
+ }
+
+ @Test
+ public void testSignedThenEncryptedAttachmentWithoutLoss() throws
Exception {
+ sendAndAssertReceived("sign-encrypt", "Signature Encrypt", false,
false);
+ }
+
+ @Test
+ public void testSignedThenEncryptedAttachmentRetransmittedTwice() throws
Exception {
+ REJECTIONS.set(1);
+ sendAndAssertReceived("sign-encrypt", "Signature Encrypt", false,
true);
+ }
+
+ @Test
+ public void
testSignedThenEncryptedAttachmentRetransmittedTwiceAfterRecovery() throws
Exception {
+ REJECTIONS.set(1);
+ sendRecoverAndAssertReceived("sign-encrypt", "Signature Encrypt",
false);
+ }
+
+ @Test
+ public void testEncryptedThenSignedAttachmentWithoutLoss() throws
Exception {
+ sendAndAssertReceived("encrypt-sign", "Encrypt Signature", false,
false);
+ }
+
+ @Test
+ public void testEncryptedThenSignedAttachmentRetransmittedTwice() throws
Exception {
+ REJECTIONS.set(1);
+ sendAndAssertReceived("encrypt-sign", "Encrypt Signature", false,
true);
+ }
+
+ @Test
+ public void
testEncryptedThenSignedAttachmentRetransmittedTwiceAfterRecovery() throws
Exception {
+ REJECTIONS.set(1);
+ sendRecoverAndAssertReceived("encrypt-sign", "Encrypt Signature",
false);
+ }
+
+ @Test
+ public void testStaxSignedThenEncryptedAttachmentWithoutLoss() throws
Exception {
+ sendAndAssertReceived("sign-encrypt", "Signature Encrypt", true,
false);
+ }
+
+ @Test
+ public void testStaxSignedThenEncryptedAttachmentRetransmittedTwice()
throws Exception {
+ REJECTIONS.set(1);
+ sendAndAssertReceived("sign-encrypt", "Signature Encrypt", true, true);
+ }
+
+ @Test
+ public void
testStaxSignedThenEncryptedAttachmentRetransmittedTwiceAfterRecovery() throws
Exception {
+ REJECTIONS.set(1);
+ sendRecoverAndAssertReceived("sign-encrypt", "Signature Encrypt",
true);
+ }
+
+ @Test
+ public void testStaxEncryptedThenSignedAttachmentWithoutLoss() throws
Exception {
+ sendAndAssertReceived("encrypt-sign", "Encrypt Signature", true,
false);
+ }
+
+ @Test
+ public void testStaxEncryptedThenSignedAttachmentRetransmittedTwice()
throws Exception {
+ REJECTIONS.set(1);
+ sendAndAssertReceived("encrypt-sign", "Encrypt Signature", true, true);
+ }
+
+ @Test
+ public void
testStaxEncryptedThenSignedAttachmentRetransmittedTwiceAfterRecovery() throws
Exception {
+ REJECTIONS.set(1);
+ sendRecoverAndAssertReceived("encrypt-sign", "Encrypt Signature",
true);
+ }
+
+ /** Sends a message with an attachment, whose first transmission is lost
if loseFirst is set */
+ private void sendAndAssertReceived(String path, String action, boolean
stax, boolean loseFirst)
+ throws Exception {
+ bus = new SpringBusFactory().createBus(RM_CONFIG);
+ initBus(2000L, loseFirst);
+ sendAndAssertReceived(path, action, stax);
+ }
+
+ private void sendAndAssertReceived(String path, String action, boolean
stax) throws Exception {
+ Greeter greeter = createGreeter(path);
+ Client client = ClientProxy.getClient(greeter);
+ client.getOutInterceptors().add(new AttachmentAdder());
+ if (action != null) {
+ client.getOutInterceptors().add(createSecurityInterceptor(action,
stax));
+ }
+
+ greeter.greetMeOneWay("CXF-6955");
+
+ assertAttachmentReceived(15);
+ }
+
+ /**
+ * Sends a message with an attachment whose transmission is lost, then
restarts the client, which
+ * recovers the message from the persistent store and retransmits it.
+ */
+ private void sendRecoverAndAssertReceived(String path, String action,
boolean stax) throws Exception {
+ String dbName = DB_NAME + "-" + DB_COUNT.getAndIncrement();
+
+ // the first client loses the message, which is persisted with its
attachment
+ createPersistentBus(dbName, 60000L, true);
+ Greeter greeter = createGreeter(path);
+ Client client = ClientProxy.getClient(greeter);
+ client.getOutInterceptors().add(new AttachmentAdder());
+ // on the endpoint, as the chain of a recovered message is built from
the interceptors of the endpoint
+
client.getEndpoint().getOutInterceptors().add(createRecoverySecurityInterceptor(action,
stax));
+ greeter.greetMeOneWay("CXF-6955");
+ assertTrue("the message must not be received before the restart",
RECEIVED.isEmpty());
+ bus.shutdown(true);
+ bus = null;
+
+ // the second client recovers the message from the store and
retransmits it
+ createPersistentBus(dbName, 2000L, false);
+ // Force recovered attachments onto disk, where closing a reader must
not delete the retry data.
+ bus.setProperty(AttachmentDeserializer.ATTACHMENT_MEMORY_THRESHOLD, 1);
+ greeter = createGreeter(path);
+ ClientProxy.getClient(greeter).getEndpoint().getOutInterceptors()
+ .add(createRecoverySecurityInterceptor(action, stax));
+
+ assertAttachmentReceived(20);
+ }
+
+ private void createPersistentBus(String dbName, long
retransmissionInterval, boolean lose) {
+ System.setProperty("db.name", dbName);
+ try {
+ bus = new SpringBusFactory().createBus(PERSISTENT_CONFIG);
+ } finally {
+ System.clearProperty("db.name");
+ }
+ initBus(retransmissionInterval, lose);
+ }
+
+ private void initBus(long retransmissionInterval, boolean lose) {
+ BusFactory.setDefaultBus(bus);
+ BusFactory.setThreadDefaultBus(bus);
+
bus.getExtension(RMManager.class).getConfiguration().setBaseRetransmissionInterval(retransmissionInterval);
+ if (lose) {
+ MessageLossSimulator loser = new MessageLossSimulator();
+ loser.setMode(-1);
+ bus.getOutInterceptors().add(loser);
+ }
+ }
+
+ private static Greeter createGreeter(String path) {
+ Greeter greeter = new GreeterService().getGreeterPort();
+ // also updates the address of the endpoint, to which recovered
messages are retransmitted
+ TestUtil.setAddress(greeter, "http://localhost:"; + PORT + "/" + path);
+ return greeter;
+ }
+
+ private static Interceptor<? extends Message>
createSecurityInterceptor(String action, boolean stax) {
+ Map<String, Object> props = createSecurityProperties(action);
+ return stax ? new WSS4JStaxOutInterceptor(props) : new
WSS4JOutInterceptor(props);
+ }
+
+ private static Interceptor<? extends Message>
createRecoverySecurityInterceptor(String action, boolean stax) {
+ return stax ? new
AttachmentsWSS4JStaxOutInterceptor(createSecurityProperties(action))
+ : createSecurityInterceptor(action, false);
+ }
+
+ static Map<String, Object> createSecurityProperties(String action) {
+ Map<String, Object> props = new HashMap<>();
+ props.put("action", action);
+ props.put("encryptionUser", "bob");
+ props.put("encryptionPropFile", "bob.properties");
+ props.put("encryptionParts", "{Content}cid:Attachments;");
+ props.put("signatureUser", "alice");
+ props.put("signaturePropFile", "alice.properties");
+ props.put("signatureParts",
"{}{http://schemas.xmlsoap.org/soap/envelope/}Body;{Content}cid:Attachments;";);
+ props.put("signatureKeyIdentifier", "DirectReference");
+ props.put("passwordCallbackClass", UTPasswordCallback.class.getName());
+ return props;
+ }
+
+ private static void assertAttachmentReceived(int timeoutSeconds) throws
InterruptedException {
+ byte[] received = RECEIVED.poll(timeoutSeconds, TimeUnit.SECONDS);
+ assertNotNull("no attachment was received", received);
+ assertArrayEquals("received \"" + new String(received,
StandardCharsets.UTF_8) + "\"", ATTACHMENT, received);
+ assertTrue("exactly one attachment is expected", RECEIVED.isEmpty());
+ }
+}
diff --git
a/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/sec/WSRMWithSecuredMtomTest.java
b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/sec/WSRMWithSecuredMtomTest.java
new file mode 100644
index 00000000000..16769301884
--- /dev/null
+++
b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/sec/WSRMWithSecuredMtomTest.java
@@ -0,0 +1,334 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.cxf.systest.ws.rm.sec;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.activation.DataHandler;
+import jakarta.jws.Oneway;
+import jakarta.jws.WebParam;
+import jakarta.jws.WebService;
+import jakarta.jws.soap.SOAPBinding;
+import jakarta.mail.util.ByteArrayDataSource;
+import jakarta.xml.bind.annotation.XmlMimeType;
+import jakarta.xml.bind.annotation.XmlRootElement;
+import jakarta.xml.ws.BindingProvider;
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
+import org.apache.cxf.attachment.AttachmentDeserializer;
+import org.apache.cxf.binding.soap.SoapMessage;
+import org.apache.cxf.bus.spring.SpringBusFactory;
+import org.apache.cxf.frontend.ClientProxy;
+import org.apache.cxf.interceptor.Fault;
+import org.apache.cxf.interceptor.Interceptor;
+import org.apache.cxf.jaxws.EndpointImpl;
+import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.AbstractPhaseInterceptor;
+import org.apache.cxf.phase.Phase;
+import org.apache.cxf.systest.ws.rm.MessageLossSimulator;
+import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
+import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
+import org.apache.cxf.ws.addressing.AddressingProperties;
+import org.apache.cxf.ws.rm.RMContextUtils;
+import org.apache.cxf.ws.rm.RMManager;
+import org.apache.cxf.ws.security.wss4j.WSS4JInInterceptor;
+import org.apache.cxf.ws.security.wss4j.WSS4JOutInterceptor;
+import org.apache.cxf.ws.security.wss4j.WSS4JStaxOutInterceptor;
+import org.apache.wss4j.common.ConfigurationConstants;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/** Exercises attachments created by JAXB marshalling, rather than
pre-existing SwA attachments. */
+@RunWith(Parameterized.class)
+public class WSRMWithSecuredMtomTest extends AbstractBusClientServerTestBase {
+ public static final String PORT =
allocatePort(WSRMWithSecuredMtomTest.class);
+
+ private static final String NAMESPACE = "urn:cxf:rm:secured-mtom";
+ private static final String RM_CONFIG =
"/org/apache/cxf/systest/ws/rm/simple.xml";
+ private static final String PERSISTENT_CONFIG =
"/org/apache/cxf/systest/ws/rm/persistent.xml";
+ private static final byte[] DATA = "MTOM attachment content of
CXF-6955".getBytes(StandardCharsets.UTF_8);
+ private static final BlockingQueue<byte[]> RECEIVED = new
LinkedBlockingQueue<>();
+ private static final BlockingQueue<String> CONTENT_TYPES = new
LinkedBlockingQueue<>();
+ private static final AtomicInteger REJECTIONS = new AtomicInteger();
+ private static final AtomicInteger DATABASES = new AtomicInteger();
+
+ private final boolean stax;
+ private final String action;
+ private Bus bus;
+
+ public WSRMWithSecuredMtomTest(boolean stax, String action) {
+ this.stax = stax;
+ this.action = action;
+ }
+
+ @Parameterized.Parameters(name = "stax={0}, action={1}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(new Object[][] {
+ {false, "Encrypt"}, {true, "Encrypt"},
+ {false, "Signature"}, {true, "Signature"},
+ {false, "Signature Encrypt"}, {true, "Signature Encrypt"},
+ {false, "Encrypt Signature"}, {true, "Encrypt Signature"}
+ });
+ }
+
+ @XmlRootElement(name = "upload", namespace = NAMESPACE)
+ public static class Upload {
+ private DataHandler data;
+
+ @XmlMimeType("application/octet-stream")
+ public DataHandler getData() {
+ return data;
+ }
+
+ public void setData(DataHandler data) {
+ this.data = data;
+ }
+ }
+
+ @WebService(targetNamespace = NAMESPACE)
+ @SOAPBinding(parameterStyle = SOAPBinding.ParameterStyle.BARE)
+ public interface Receiver {
+ @Oneway
+ void upload(@WebParam(name = "upload", targetNamespace = NAMESPACE)
Upload upload);
+ }
+
+ @WebService(endpointInterface =
"org.apache.cxf.systest.ws.rm.sec.WSRMWithSecuredMtomTest$Receiver",
+ targetNamespace = NAMESPACE)
+ public static class ReceiverImpl implements Receiver {
+ public void upload(Upload upload) {
+ try (InputStream in = upload.getData().getInputStream()) {
+ RECEIVED.add(in.readAllBytes());
+ } catch (Exception e) {
+ throw new Fault(e);
+ }
+ }
+ }
+
+ public static class Server extends AbstractBusTestServerBase {
+ protected void run() {
+ Bus serverBus = new SpringBusFactory().createBus(RM_CONFIG);
+ BusFactory.setDefaultBus(serverBus);
+ setBus(serverBus);
+ for (String action : new String[] {"Encrypt", "Signature",
"Signature Encrypt", "Encrypt Signature"}) {
+ EndpointImpl endpoint = new EndpointImpl(serverBus, new
ReceiverImpl());
+ endpoint.getProperties().put(Message.MTOM_ENABLED,
Boolean.TRUE);
+ endpoint.getInInterceptors().add(new
AbstractPhaseInterceptor<Message>(Phase.RECEIVE) {
+ public void handleMessage(Message message) {
+ String type =
(String)message.get(Message.CONTENT_TYPE);
+ if (type != null && type.startsWith("multipart/")) {
+ if (REJECTIONS.getAndUpdate(n -> Math.max(0, n -
1)) > 0) {
+ throw new Fault(new
IllegalStateException("simulated failure"));
+ }
+ CONTENT_TYPES.add(type);
+ }
+ }
+ });
+ Map<String, Object> props = new HashMap<>();
+ props.put("action", action);
+ props.put("decryptionPropFile", "bob.properties");
+ props.put("signatureVerificationPropFile", "bob.properties");
+ props.put("passwordCallbackClass",
UTPasswordCallback.class.getName());
+ props.put(ConfigurationConstants.EXPAND_XOP_INCLUDE, "false");
+ endpoint.getInInterceptors().add(new WSS4JInInterceptor(props)
{
+ @Override
+ public void handleMessage(SoapMessage message) {
+ String type =
(String)message.get(Message.CONTENT_TYPE);
+ if (type != null && type.startsWith("multipart/")) {
+ super.handleMessage(message);
+ }
+ }
+ });
+ endpoint.publish(address(action));
+ }
+ }
+ }
+
+ @BeforeClass
+ public static void startServer() throws Exception {
+ assertTrue("server did not launch", launchServer(Server.class, true));
+ }
+
+ @Before
+ public void clearReceived() {
+ RECEIVED.clear();
+ CONTENT_TYPES.clear();
+ REJECTIONS.set(0);
+ }
+
+ @After
+ public void shutdownBus() {
+ REJECTIONS.set(0);
+ if (bus != null) {
+ bus.shutdown(true);
+ bus = null;
+ }
+ }
+
+ @Test
+ public void testWithoutLoss() throws Exception {
+ createBus(null, false, 2000L);
+ send(createClient(false));
+ assertReceived();
+ }
+
+ @Test
+ public void testRetransmittedTwice() throws Exception {
+ REJECTIONS.set(1);
+ createBus(null, true, 2000L);
+ send(createClient(false));
+ assertReceived();
+ }
+
+ @Test
+ public void testPersistentStoreWithoutLoss() throws Exception {
+ createBus(database(), false, 2000L);
+ send(createClient(false));
+ assertReceived();
+ }
+
+ @Test
+ public void testRetransmittedTwiceAfterRecovery() throws Exception {
+ String database = database();
+ createBus(database, true, 60000L);
+ send(createClient(true));
+ assertTrue("the initial transmission must be lost",
RECEIVED.isEmpty());
+ shutdownBus();
+
+ REJECTIONS.set(1);
+ createBus(database, false, 2000L);
+ bus.setProperty(AttachmentDeserializer.ATTACHMENT_MEMORY_THRESHOLD, 1);
+ createClient(true);
+ assertReceived();
+ }
+
+ private static String database() {
+ return "wsrm-sec-mtom-" + DATABASES.getAndIncrement();
+ }
+
+ private static String address(String action) {
+ return "http://localhost:"; + PORT + "/" + action.replace(' ', '-');
+ }
+
+ private void createBus(String database, boolean loseFirst, long interval) {
+ if (database == null) {
+ bus = new SpringBusFactory().createBus(RM_CONFIG);
+ } else {
+ System.setProperty("db.name", database);
+ try {
+ bus = new SpringBusFactory().createBus(PERSISTENT_CONFIG);
+ } finally {
+ System.clearProperty("db.name");
+ }
+ }
+ BusFactory.setDefaultBus(bus);
+ BusFactory.setThreadDefaultBus(bus);
+
bus.getExtension(RMManager.class).getConfiguration().setBaseRetransmissionInterval(interval);
+ if (loseFirst) {
+ MessageLossSimulator loser = new MessageLossSimulator();
+ loser.setMode(-1);
+ bus.getOutInterceptors().add(loser);
+ }
+ }
+
+ private Receiver createClient(boolean recovery) {
+ JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
+ factory.setBus(bus);
+ factory.setServiceClass(Receiver.class);
+ factory.setAddress(address(action));
+ Receiver client = factory.create(Receiver.class);
+
((jakarta.xml.ws.soap.SOAPBinding)((BindingProvider)client).getBinding()).setMTOMEnabled(true);
+ Map<String, Object> props =
WSRMWithSecuredAttachmentsTest.createSecurityProperties(action);
+ // Keep the XOP reference intact and secure its attachment explicitly.
+ props.put(ConfigurationConstants.EXPAND_XOP_INCLUDE, "false");
+ props.put(ConfigurationConstants.STORE_BYTES_IN_ATTACHMENT, "false");
+ // Focus on attachment signing: streaming Body signing finalizes
external references before
+ // JAXB has created the MTOM attachments. The SwA tests also sign the
Body.
+ props.put(ConfigurationConstants.SIGNATURE_PARTS,
"{Content}cid:Attachments;");
+ Interceptor<? extends Message> security;
+ if (stax) {
+ WSS4JStaxOutInterceptor interceptor = recovery ? new
WSS4JStaxOutInterceptor(props) {
+ @Override
+ public void handleMessage(SoapMessage message) {
+ if (isApplicationMessage(message)) {
+ super.handleMessage(message);
+ }
+ }
+ } : new WSS4JStaxOutInterceptor(props);
+ interceptor.setAllowMTOM(true);
+ security = interceptor;
+ } else {
+ security = recovery ? new WSS4JOutInterceptor(props) {
+ @Override
+ public void handleMessage(SoapMessage message) {
+ if (isApplicationMessage(message)) {
+ super.handleMessage(message);
+ }
+ }
+ } : new WSS4JOutInterceptor(props);
+ }
+ if (recovery) {
+
ClientProxy.getClient(client).getEndpoint().getOutInterceptors().add(security);
+ } else {
+ ClientProxy.getClient(client).getOutInterceptors().add(security);
+ }
+ return client;
+ }
+
+ private static boolean isApplicationMessage(SoapMessage message) {
+ AddressingProperties maps = RMContextUtils.retrieveMAPs(message,
false, true);
+ return maps != null && maps.getAction() != null
+ &&
!RMContextUtils.isRMProtocolMessage(maps.getAction().getValue());
+ }
+
+ private static void send(Receiver client) {
+ Upload upload = new Upload();
+ upload.setData(new DataHandler(new ByteArrayDataSource(DATA,
"application/octet-stream")));
+ client.upload(upload);
+ }
+
+ private static void assertReceived() throws Exception {
+ byte[] received = RECEIVED.poll(20, TimeUnit.SECONDS);
+ assertNotNull("no MTOM payload was received", received);
+ assertArrayEquals(DATA, received);
+ String type = CONTENT_TYPES.poll(1, TimeUnit.SECONDS);
+ assertNotNull("the request must be multipart", type);
+ assertTrue("MTOM must remain enabled: " + type,
type.contains("application/xop+xml"));
+ assertTrue("exactly one payload is expected", RECEIVED.isEmpty());
+ }
+}
