This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch 11.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/11.0.x by this push:
new b0d35b03d7 Implement TSL-PSK encryption for clustering
b0d35b03d7 is described below
commit b0d35b03d7d89b063559c209e0a18122b13f3f02
Author: Mark Thomas <[email protected]>
AuthorDate: Tue Sep 22 17:12:46 2026 +0100
Implement TSL-PSK encryption for clustering
Co-authored-by: GitHub Copilot (GPT-5.6 Sol) <[email protected]>
---
.../apache/catalina/tribes/group/GroupChannel.java | 64 ++++++
.../catalina/tribes/group/GroupChannelMBean.java | 4 +
.../catalina/tribes/group/LocalStrings.properties | 5 +
.../catalina/tribes/group/TribesSslContext.java | 153 ++++++++++++++
.../catalina/tribes/transport/AbstractSender.java | 11 +
.../catalina/tribes/transport/ReceiverBase.java | 24 +++
.../tribes/transport/ReplicationTransmitter.java | 7 +
.../tribes/transport/nio/LocalStrings.properties | 8 +
.../catalina/tribes/transport/nio/NioReceiver.java | 148 +++++++++++++-
.../tribes/transport/nio/ParallelNioSender.java | 156 +++++++++++++++
.../catalina/tribes/transport/nio/TlsChannel.java | 222 +++++++++++++++++++++
.../apache/tomcat/jni/PreSharedKeySelector.java | 16 +-
java/org/apache/tomcat/jni/SSLContext.java | 18 +-
java/org/apache/tomcat/util/net/SSLContext.java | 13 ++
.../tomcat/util/net/openssl/OpenSSLContext.java | 25 ++-
.../net/openssl/OpenSSLPreSharedKeySelector.java | 13 +-
.../util/net/openssl/panama/OpenSSLContext.java | 59 +++++-
.../util/openssl/SSL_psk_client_cb_func.java | 44 ++++
java/org/apache/tomcat/util/openssl/openssl_h.java | 29 +++
res/openssl/openssl-tomcat.conf | 3 +
.../catalina/tribes/group/TestGroupChannelTls.java | 84 ++++++++
.../tribes/transport/nio/TestTlsChannel.java | 112 +++++++++++
webapps/docs/changelog.xml | 3 +
webapps/docs/config/cluster-channel.xml | 22 ++
24 files changed, 1226 insertions(+), 17 deletions(-)
diff --git a/java/org/apache/catalina/tribes/group/GroupChannel.java
b/java/org/apache/catalina/tribes/group/GroupChannel.java
index 99eb3be4c9..840559a48b 100644
--- a/java/org/apache/catalina/tribes/group/GroupChannel.java
+++ b/java/org/apache/catalina/tribes/group/GroupChannel.java
@@ -124,6 +124,11 @@ public class GroupChannel extends ChannelInterceptorBase
implements ManagedChann
*/
protected boolean optionCheck = false;
+ private boolean secure;
+ private String pskIdentity;
+ private String pskKey;
+ private volatile TribesSslContext sslContext;
+
/**
* the name of this channel.
*/
@@ -218,6 +223,16 @@ public class GroupChannel extends ChannelInterceptorBase
implements ManagedChann
}
XByteBuffer buffer = null;
try {
+ if (secure) {
+ options |= SEND_OPTIONS_SECURE;
+ }
+ if ((options & SEND_OPTIONS_SECURE) != 0 &&
+ (options & (SEND_OPTIONS_UDP | SEND_OPTIONS_MULTICAST)) !=
0) {
+ throw new
ChannelException(sm.getString("groupChannel.tlsDatagramUnsupported"));
+ }
+ if ((options & SEND_OPTIONS_SECURE) != 0 && sslContext == null) {
+ throw new
ChannelException(sm.getString("groupChannel.tlsUnavailable"));
+ }
if (destination == null || destination.length == 0) {
throw new
ChannelException(sm.getString("groupChannel.noDestination"));
}
@@ -443,6 +458,25 @@ public class GroupChannel extends ChannelInterceptorBase
implements ManagedChann
@Override
public synchronized void start(int svc) throws ChannelException {
setupDefaultStack();
+ if (sslContext == null) {
+ if (pskKey != null && pskIdentity != null) {
+ try {
+ sslContext = new TribesSslContext(pskIdentity, pskKey);
+ } catch (Exception e) {
+ if (secure) {
+ throw new
ChannelException(sm.getString("groupChannel.tlsUnavailable"), e);
+ }
+ log.warn(sm.getString("groupChannel.tlsUnavailable"), e);
+ }
+ } else if (secure && pskKey == null) {
+ throw new
ChannelException(sm.getString("groupChannel.tlsKeyMissing"));
+ } else if (secure && pskIdentity == null) {
+ throw new
ChannelException(sm.getString("groupChannel.tlsIdentityMissing"));
+ }
+ }
+ if (secure && getChannelReceiver().getSecurePort() < 0) {
+ throw new
ChannelException(sm.getString("groupChannel.tlsPortMissing"));
+ }
if (optionCheck) {
checkOptionFlags();
}
@@ -497,6 +531,10 @@ public class GroupChannel extends ChannelInterceptorBase
implements ManagedChann
heartbeatFuture.cancel(true);
heartbeatFuture = null;
}
+ if (sslContext != null) {
+ sslContext.close();
+ sslContext = null;
+ }
if (ownExecutor && utilityExecutor != null) {
utilityExecutor.shutdown();
utilityExecutor = null;
@@ -630,6 +668,32 @@ public class GroupChannel extends ChannelInterceptorBase
implements ManagedChann
return optionCheck;
}
+ @Override
+ public boolean getSecure() {
+ return secure;
+ }
+
+ public void setSecure(boolean secure) {
+ this.secure = secure;
+ }
+
+ @Override
+ public String getPskIdentity() {
+ return pskIdentity;
+ }
+
+ public void setPskIdentity(String pskIdentity) {
+ this.pskIdentity = pskIdentity;
+ }
+
+ public void setPskKey(String pskKey) {
+ this.pskKey = pskKey;
+ }
+
+ public TribesSslContext getSslContext() {
+ return sslContext;
+ }
+
@Override
public boolean getHeartbeat() {
return heartbeat;
diff --git a/java/org/apache/catalina/tribes/group/GroupChannelMBean.java
b/java/org/apache/catalina/tribes/group/GroupChannelMBean.java
index 74dbf6da64..bd63930a9c 100644
--- a/java/org/apache/catalina/tribes/group/GroupChannelMBean.java
+++ b/java/org/apache/catalina/tribes/group/GroupChannelMBean.java
@@ -52,6 +52,10 @@ public interface GroupChannelMBean {
*/
long getHeartbeatSleeptime();
+ boolean getSecure();
+
+ String getPskIdentity();
+
// Operations
/**
* Starts the channel with the given service type.
diff --git a/java/org/apache/catalina/tribes/group/LocalStrings.properties
b/java/org/apache/catalina/tribes/group/LocalStrings.properties
index 422fced1f4..c348f8b2c2 100644
--- a/java/org/apache/catalina/tribes/group/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/group/LocalStrings.properties
@@ -23,6 +23,11 @@ channelCoordinator.invalidState.notStopped=Configuration may
not be changed unti
groupChannel.listener.alreadyExist=Listener already exists:[{0}][{1}]
groupChannel.noDestination=No destination given
groupChannel.nullMessage=Cannot send a NULL message
+groupChannel.tlsIdentityMissing=The channel is configured as secure but no TLS
pre-shared key identity is configured
+groupChannel.tlsKeyMissing=The channel is configured as secure but no TLS
pre-shared key is configured
+groupChannel.tlsDatagramUnsupported=Secure Tribes messages cannot use UDP or
multicast
+groupChannel.tlsPortMissing=The channel is configured as secure but the
receiver has no secure port configured
+groupChannel.tlsUnavailable=TLS is not available for this channel
groupChannel.optionFlag.conflict=Interceptor option flag conflict: [{0}]
groupChannel.receiving.error=Error receiving message:
groupChannel.sendFail.noRpcChannelReply=Unable to find rpc channel, failed to
send NoRpcChannelReply.
diff --git a/java/org/apache/catalina/tribes/group/TribesSslContext.java
b/java/org/apache/catalina/tribes/group/TribesSslContext.java
new file mode 100644
index 0000000000..d302bdfc34
--- /dev/null
+++ b/java/org/apache/catalina/tribes/group/TribesSslContext.java
@@ -0,0 +1,153 @@
+/*
+ * 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.catalina.tribes.group;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Method;
+import java.util.Set;
+
+import javax.net.ssl.SSLEngine;
+
+/**
+ * The pair (client and server) of TLS contexts used by a Tribes channel.
+ */
+public class TribesSslContext implements AutoCloseable {
+
+ private static final String FFM_CONTEXT =
"org.apache.tomcat.util.net.openssl.panama.OpenSSLContext";
+ private static final String NATIVE_CONTEXT =
"org.apache.tomcat.util.net.openssl.OpenSSLContext";
+
+ private final Object clientContext;
+ private final Object serverContext;
+ private final boolean ffm;
+
+ public TribesSslContext(String identity, String key) throws Exception {
+ Object[] contexts = createFfmContexts(identity, key);
+ if (contexts != null) {
+ ffm = true;
+ } else {
+ contexts = createNativeContexts(identity, key);
+ ffm = false;
+ }
+ if (contexts == null) {
+ throw new IllegalStateException("Neither OpenSSL FFM nor Tomcat
Native is available");
+ }
+ clientContext = contexts[0];
+ serverContext = contexts[1];
+ }
+
+ public SSLEngine createClientEngine() {
+ return createEngine(clientContext, true);
+ }
+
+ public SSLEngine createServerEngine() {
+ return createEngine(serverContext, false);
+ }
+
+ private static Object[] createFfmContexts(String identity, String key)
throws Exception {
+ /*
+ * Tribes may be used stand-alone so don't assume there is an
OpenSSLLifecycleListener that Tomcat is already
+ * using but do use the library in a manner that is compatible if
Tomcat is using such an instance.
+ */
+ Class<?> library;
+ try {
+ library =
Class.forName("org.apache.tomcat.util.net.openssl.panama.OpenSSLLibrary");
+ library.getMethod("init").invoke(null);
+ } catch (Throwable t) {
+ return null;
+ }
+ Class<?> status =
Class.forName("org.apache.tomcat.util.net.openssl.OpenSSLStatus");
+ if (!((Boolean)
status.getMethod("isAvailable").invoke(null)).booleanValue()) {
+ library.getMethod("destroy").invoke(null);
+ return null;
+ }
+ return createContexts(FFM_CONTEXT, identity, key);
+ }
+
+ private static Object[] createNativeContexts(String identity, String key)
throws Exception {
+ /*
+ * Tribes may be used stand-alone so don't assume there is an
AprLifecycleListener that Tomcat is already using
+ * but do use the library in a manner that is compatible if Tomcat is
using such an instance.
+ */
+ Class<?> listener =
Class.forName("org.apache.catalina.core.AprLifecycleListener");
+ listener.getConstructor().newInstance();
+ if (!((Boolean)
listener.getMethod("isAprAvailable").invoke(null)).booleanValue()) {
+ return null;
+ }
+ return createContexts(NATIVE_CONTEXT, identity, key);
+ }
+
+ private static Object[] createContexts(String className, String identity,
String key) throws Exception {
+ Class<?> clazz = Class.forName(className);
+ Class<?> certificateClass =
Class.forName("org.apache.tomcat.util.net.SSLHostConfigCertificate");
+ Constructor<?> constructor = clazz.getConstructor(certificateClass,
java.util.List.class, boolean.class);
+ Object client = constructor.newInstance(createCertificate(identity,
key), null, Boolean.TRUE);
+ Object server = constructor.newInstance(createCertificate(identity,
key), null, Boolean.FALSE);
+ Method init = clazz.getMethod("init",
javax.net.ssl.KeyManager[].class, javax.net.ssl.TrustManager[].class,
+ java.security.SecureRandom.class);
+ init.invoke(client, null, null, null);
+ init.invoke(server, null, null, null);
+ return new Object[] { client, server };
+ }
+
+ private static Object createCertificate(String identity, String key)
throws Exception {
+ Class<?> configClass =
Class.forName("org.apache.tomcat.util.net.SSLHostConfig");
+ Object config = configClass.getConstructor().newInstance();
+ configClass.getMethod("setProtocols", String.class).invoke(config,
"TLSv1.2");
+ configClass.getMethod("setEnabledProtocols",
String[].class).invoke(config,
+ (Object) new String[] { "TLSv1.2" });
+ configClass.getMethod("setCiphers", String.class).invoke(config,
"PSK-AES128-GCM-SHA256");
+ Class<?> pskClass =
Class.forName("org.apache.tomcat.util.net.SSLHostConfigPreSharedKey");
+ Object psk = pskClass.getConstructor(configClass).newInstance(config);
+ pskClass.getMethod("setIdentity", String.class).invoke(psk, identity);
+ pskClass.getMethod("setKey", String.class).invoke(psk, key);
+ configClass.getMethod("addPreSharedKey", pskClass).invoke(config, psk);
+ Set<?> certificates =
+ (Set<?>) configClass.getMethod("getCertificates",
boolean.class).invoke(config, Boolean.TRUE);
+ return certificates.iterator().next();
+ }
+
+ private static SSLEngine createEngine(Object context, boolean clientMode) {
+ try {
+ return (SSLEngine) context.getClass().getMethod("createSSLEngine",
boolean.class).invoke(context,
+ Boolean.valueOf(clientMode));
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Override
+ public void close() {
+ try {
+
clientContext.getClass().getMethod("destroy").invoke(clientContext);
+ } catch (ReflectiveOperationException e) {
+ // The contexts may already have been destroyed during channel
shutdown.
+ }
+ try {
+
serverContext.getClass().getMethod("destroy").invoke(serverContext);
+ } catch (ReflectiveOperationException e) {
+ // The contexts may already have been destroyed during channel
shutdown.
+ }
+ if (ffm) {
+ try {
+
Class.forName("org.apache.tomcat.util.net.openssl.panama.OpenSSLLibrary").getMethod("destroy")
+ .invoke(null);
+ } catch (ReflectiveOperationException e) {
+ // The contexts have already released their native resources.
+ }
+ }
+ }
+}
diff --git a/java/org/apache/catalina/tribes/transport/AbstractSender.java
b/java/org/apache/catalina/tribes/transport/AbstractSender.java
index bbe3dbcbab..aa899e62ff 100644
--- a/java/org/apache/catalina/tribes/transport/AbstractSender.java
+++ b/java/org/apache/catalina/tribes/transport/AbstractSender.java
@@ -20,6 +20,7 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.TribesSslContext;
/** Abstract base implementation of a data sender. */
public abstract class AbstractSender implements DataSender {
@@ -76,6 +77,7 @@ public abstract class AbstractSender implements DataSender {
private boolean udpBased = false;
/** The UDP port of the destination. */
private int udpPort = -1;
+ private TribesSslContext sslContext;
/**
* transfers sender properties from one sender to another
@@ -106,6 +108,7 @@ public abstract class AbstractSender implements DataSender {
to.throwOnFailedAck = from.throwOnFailedAck;
to.udpBased = from.udpBased;
to.udpPort = from.udpPort;
+ to.sslContext = from.sslContext;
}
@@ -220,6 +223,14 @@ public abstract class AbstractSender implements DataSender
{
return port;
}
+ public TribesSslContext getSslContext() {
+ return sslContext;
+ }
+
+ public void setSslContext(TribesSslContext sslContext) {
+ this.sslContext = sslContext;
+ }
+
/**
* Returns the maximum number of retry attempts.
* @return the maximum retry attempts
diff --git a/java/org/apache/catalina/tribes/transport/ReceiverBase.java
b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
index 60594c0665..0595b28e59 100644
--- a/java/org/apache/catalina/tribes/transport/ReceiverBase.java
+++ b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
@@ -267,6 +267,30 @@ public abstract class ReceiverBase implements
ChannelReceiver, ListenCallback, R
log.info(sm.getString("receiverBase.unable.bind",
addr));
throw ioe;
}
+
+ port++;
+ }
+
+ }
+ }
+ }
+
+ protected void bindSecure(ServerSocket socket, int portstart, int retries)
throws IOException {
+ synchronized (bindLock) {
+ InetSocketAddress address = null;
+ int port = portstart;
+ while (retries > 0) {
+ try {
+ address = new InetSocketAddress(getBind(), port);
+ socket.bind(address);
+ setSecurePort(port);
+ log.info(sm.getString("receiverBase.socket.bind",
address));
+ return;
+ } catch (IOException ioe) {
+ if (--retries <= 0) {
+ log.info(sm.getString("receiverBase.unable.bind",
address));
+ throw ioe;
+ }
port++;
}
}
diff --git
a/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
b/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
index 64a729ef07..7a58edf22f 100644
--- a/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
+++ b/java/org/apache/catalina/tribes/transport/ReplicationTransmitter.java
@@ -25,6 +25,7 @@ import org.apache.catalina.tribes.ChannelException;
import org.apache.catalina.tribes.ChannelMessage;
import org.apache.catalina.tribes.ChannelSender;
import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.GroupChannel;
import org.apache.catalina.tribes.jmx.JmxRegistry;
import org.apache.catalina.tribes.transport.nio.PooledParallelSender;
@@ -70,6 +71,9 @@ public class ReplicationTransmitter implements ChannelSender {
@Override
public void sendMessage(ChannelMessage message, Member[] destination)
throws ChannelException {
+ if (channel instanceof GroupChannel groupChannel &&
groupChannel.getSecure()) {
+ message.setOptions(message.getOptions() |
Channel.SEND_OPTIONS_SECURE);
+ }
MultiPointSender sender = getTransport();
sender.sendMessage(destination, message);
}
@@ -148,6 +152,9 @@ public class ReplicationTransmitter implements
ChannelSender {
@Override
public void setChannel(Channel channel) {
this.channel = channel;
+ if (transport instanceof AbstractSender sender && channel instanceof
GroupChannel groupChannel) {
+ sender.setSslContext(groupChannel.getSslContext());
+ }
}
}
diff --git
a/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
b/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
index 1c5a55d22b..4308acbacc 100644
--- a/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
+++ b/java/org/apache/catalina/tribes/transport/nio/LocalStrings.properties
@@ -25,9 +25,12 @@ nioReceiver.noThread=No TcpReplicationThread available
nioReceiver.requestError=Unable to process request in NioReceiver
nioReceiver.run.fail=Unable to run replication listener
nioReceiver.start.fail=Unable to start cluster receiver
+nioReceiver.stop.executor.interrupted=The NioReceiver executor was interrupted
while shutting down
+nioReceiver.stop.executor.timeout=The NioReceiver executor did not shutdown
within the allowed timeout
nioReceiver.stop.fail=Unable to close cluster receiver selector
nioReceiver.stop.threadRunning=The NioReceiver thread did not stop in a timely
manner. Errors may be observed when the selector is closed.
nioReceiver.threadpool.fail=ThreadPool cannot be initialized. Listener not
started.
+nioReceiver.tlsUnavailable=A secure port is configured but TLS is not available
nioReceiver.threadsExhausted=Channel key is registered, but has had no
interest ops for the last [{0}] ms. (cancelled: [{1}]):[{2}] last access:[{3}]
Possible cause: all threads used, perform thread dump
nioReplicationTask.disconnect=Channel closed on the remote end, disconnecting
@@ -48,8 +51,13 @@ nioSender.unable.disconnect=Unable to disconnect NioSender.
msg=[{0}]
nioSender.unable.receive.ack=Unable to receive an ack message. EOF on socket
channel has been reached.
nioSender.unknown.state=Data is in unknown state. readyOps=[{0}]
+parallelNioSender.disconnect.executor.interrupted=The ParallelNioSender
executor was interrupted while shutting down
+parallelNioSender.disconnect.executor.timeout=The ParallelNioSender executor
did not shutdown within the allowed timeout
parallelNioSender.error.keepalive=Error during keepalive test for sender:[{0}]
+parallelNioSender.invalidAck=The secure sender received an invalid
acknowledgement
parallelNioSender.operation.timedout=Operation has timed out([{0}] ms.).
+parallelNioSender.securePortUnavailable=The destination does not advertise a
secure Tribes port
+parallelNioSender.tlsUnavailable=TLS is not available for this sender
parallelNioSender.selectorCloseFail=Failed to close selector
parallelNioSender.send.fail=Member send is failing for:[{0}] ; Setting to
suspect.
parallelNioSender.send.fail.retrying=Member send is failing for:[{0}] ;
Setting to suspect and retrying.
diff --git a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
index acaa25742b..7da08a7250 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
@@ -18,6 +18,8 @@ package org.apache.catalina.tribes.transport.nio;
import java.io.IOException;
import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.DatagramChannel;
@@ -29,11 +31,19 @@ import java.nio.channels.SocketChannel;
import java.util.Deque;
import java.util.Iterator;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.group.TribesSslContext;
+import org.apache.catalina.tribes.io.ChannelData;
import org.apache.catalina.tribes.io.ObjectReader;
import org.apache.catalina.tribes.transport.AbstractRxTask;
+import org.apache.catalina.tribes.transport.Constants;
import org.apache.catalina.tribes.transport.ReceiverBase;
import org.apache.catalina.tribes.transport.RxTaskPool;
import org.apache.catalina.tribes.util.ExceptionUtils;
@@ -57,7 +67,10 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
private final AtomicReference<Selector> selector = new AtomicReference<>();
private ServerSocketChannel serverChannel = null;
+ private volatile ServerSocketChannel secureServerChannel = null;
private DatagramChannel datagramChannel = null;
+ private final Set<SocketChannel> secureSockets =
ConcurrentHashMap.newKeySet();
+ private volatile Semaphore secureConnectionSlots;
/**
* Queue of events to be processed by the selector thread.
@@ -70,10 +83,134 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
public NioReceiver() {
}
+ private void startSecureListener() throws IOException {
+ if (getSecurePort() < 0) {
+ return;
+ }
+ if (!(getChannel() instanceof GroupChannel groupChannel)) {
+ return;
+ }
+ if (groupChannel.getSslContext() == null) {
+ if (groupChannel.getSecure()) {
+ throw new
IOException(sm.getString("nioReceiver.tlsUnavailable"));
+ }
+ setSecurePort(-1);
+ return;
+ }
+ secureServerChannel = ServerSocketChannel.open();
+ bindSecure(secureServerChannel.socket(), getSecurePort(),
getAutoBind());
+ secureConnectionSlots = new Semaphore(Math.max(1, getMaxTasks()));
+ Thread thread = new Thread(() ->
runSecureListener(groupChannel.getSslContext()), "NioReceiver-TLS");
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ private void runSecureListener(TribesSslContext sslContext) {
+ ServerSocketChannel server = secureServerChannel;
+ while (server != null && server.isOpen()) {
+ try {
+ SocketChannel socket = server.accept();
+ socket.configureBlocking(true);
+ socket.socket().setSoTimeout(getTimeout());
+ Semaphore connectionSlots = secureConnectionSlots;
+ if (connectionSlots == null || !connectionSlots.tryAcquire()) {
+ socket.close();
+ continue;
+ }
+ secureSockets.add(socket);
+ try {
+ getExecutor().execute(() -> {
+ try {
+ processSecureConnection(socket, sslContext);
+ } finally {
+ secureSockets.remove(socket);
+ connectionSlots.release();
+ }
+ });
+ } catch (RuntimeException re) {
+ secureSockets.remove(socket);
+ connectionSlots.release();
+ socket.close();
+ log.warn(sm.getString("nioReceiver.requestError"), re);
+ }
+ } catch (IOException ioe) {
+ if (isListening()) {
+ log.warn(sm.getString("nioReceiver.requestError"), ioe);
+ }
+ }
+ }
+ }
+
+ private void processSecureConnection(SocketChannel socketChannel,
TribesSslContext sslContext) {
+ try (Socket socket = socketChannel.socket()) {
+ try (TlsChannel channel = new TlsChannel(socket,
sslContext.createServerEngine())) {
+ ObjectReader reader = new ObjectReader(getRxBufSize());
+ ByteBuffer buffer = ByteBuffer.allocate(getRxBufSize());
+ while (channel.isOpen()) {
+ int read = channel.read(buffer);
+ if (read < 0) {
+ return;
+ }
+ buffer.flip();
+ reader.append(buffer, read, false);
+ buffer.clear();
+ for (ChannelMessage message : reader.execute()) {
+ if (ChannelData.sendAckAsync(message.getOptions())) {
+
channel.write(ByteBuffer.wrap(Constants.ACK_COMMAND));
+ }
+ try {
+ messageDataReceived(message);
+ if (ChannelData.sendAckSync(message.getOptions()))
{
+
channel.write(ByteBuffer.wrap(Constants.ACK_COMMAND));
+ }
+ } catch (RuntimeException re) {
+ if (ChannelData.sendAckSync(message.getOptions()))
{
+
channel.write(ByteBuffer.wrap(Constants.FAIL_ACK_COMMAND));
+ }
+ }
+ }
+ }
+ }
+ } catch (IOException ioe) {
+ if (log.isDebugEnabled()) {
+ log.debug(sm.getString("nioReceiver.requestError"), ioe);
+ }
+ } catch (RuntimeException re) {
+ log.warn(sm.getString("nioReceiver.requestError"), re);
+ }
+ }
+
@Override
public void stop() {
this.stopListening();
+ if (secureServerChannel != null) {
+ try {
+ secureServerChannel.close();
+ } catch (IOException ioe) {
+ log.debug(sm.getString("nioReceiver.closeError"), ioe);
+ }
+ secureServerChannel = null;
+ }
+ for (SocketChannel socket : secureSockets) {
+ try {
+ socket.close();
+ } catch (IOException ioe) {
+ log.debug(sm.getString("nioReceiver.closeError"), ioe);
+ }
+ }
+ secureSockets.clear();
+ secureConnectionSlots = null;
super.stop();
+ if ((getChannel() instanceof GroupChannel groupChannel) &&
groupChannel.getSecure()) {
+ try {
+ // Should stop a lot faster than this as all the sockets have
been closed.
+ if (!getExecutor().awaitTermination(60, TimeUnit.SECONDS)) {
+
log.warn(sm.getString("nioReceiver.stop.executor.timeout"));
+ }
+ } catch (InterruptedException e) {
+
log.warn(sm.getString("nioReceiver.stop.executor.interrupted"));
+ }
+ }
}
@Override
@@ -91,7 +228,16 @@ public class NioReceiver extends ReceiverBase implements
Runnable, NioReceiverMB
}
try {
resolveBind();
- bind();
+ getBind();
+ boolean tlsOnly = getChannel() instanceof GroupChannel
groupChannel && groupChannel.getSecure();
+ if (!tlsOnly) {
+ bind();
+ }
+ startSecureListener();
+ if (tlsOnly) {
+ setListen(true);
+ return;
+ }
String channelName = "";
if (getChannel().getName() != null) {
channelName = "[" + getChannel().getName() + "]";
diff --git
a/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
b/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
index 9663df083f..11876aae08 100644
--- a/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
+++ b/java/org/apache/catalina/tribes/transport/nio/ParallelNioSender.java
@@ -18,20 +18,30 @@ package org.apache.catalina.tribes.transport.nio;
import java.io.IOException;
import java.lang.ref.Cleaner;
+import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import org.apache.catalina.tribes.Channel;
import org.apache.catalina.tribes.ChannelException;
import org.apache.catalina.tribes.ChannelMessage;
import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.RemoteProcessException;
import org.apache.catalina.tribes.UniqueId;
import org.apache.catalina.tribes.io.ChannelData;
import org.apache.catalina.tribes.io.XByteBuffer;
@@ -58,6 +68,8 @@ public class ParallelNioSender extends AbstractSender
implements MultiPointSende
private final InternalState state;
+ private ExecutorService sendSecureExecutor;
+
/**
* The timeout in milliseconds for the selector select operation.
*/
@@ -78,6 +90,10 @@ public class ParallelNioSender extends AbstractSender
implements MultiPointSende
@Override
public synchronized void sendMessage(Member[] destination, ChannelMessage
msg) throws ChannelException {
long start = System.currentTimeMillis();
+ if ((msg.getOptions() & Channel.SEND_OPTIONS_SECURE) != 0) {
+ sendSecure(destination, msg);
+ return;
+ }
this.setUdpBased((msg.getOptions() & Channel.SEND_OPTIONS_UDP) ==
Channel.SEND_OPTIONS_UDP);
byte[] data = XByteBuffer.createDataPackage((ChannelData) msg);
NioSender[] senders = setupForSend(destination);
@@ -155,6 +171,133 @@ public class ParallelNioSender extends AbstractSender
implements MultiPointSende
}
+ private void sendSecure(Member[] destination, ChannelMessage msg) throws
ChannelException {
+ if (getSslContext() == null) {
+ throw new
ChannelException(sm.getString("parallelNioSender.tlsUnavailable"));
+ }
+ byte[] data = XByteBuffer.createDataPackage((ChannelData) msg);
+ boolean waitForAck = (msg.getOptions() & Channel.SEND_OPTIONS_USE_ACK)
!= 0;
+ long deadline = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(getTimeout());
+ ChannelException failure = null;
+ if (destination.length == 0) {
+ return;
+ }
+ if (sendSecureExecutor == null) {
+ sendSecureExecutor = new ThreadPoolExecutor(1,
Runtime.getRuntime().availableProcessors(), 60L,
+ TimeUnit.SECONDS, new SynchronousQueue<>(), runnable -> {
+ Thread thread = new Thread(runnable,
"Tribes-TLS-Sender");
+ return thread;
+ });
+ }
+ List<Map.Entry<Member,Future<Void>>> sends = new
ArrayList<>(destination.length);
+ for (Member member : destination) {
+ Future<Void> send = sendSecureExecutor.submit(() -> {
+ sendSecure(member, data, waitForAck, deadline);
+ return null;
+ });
+ sends.add(Map.entry(member, send));
+ }
+ for (Map.Entry<Member,Future<Void>> entry : sends) {
+ try {
+ long remaining = deadline - System.nanoTime();
+ entry.getValue().get(Math.max(0, remaining),
TimeUnit.NANOSECONDS);
+ } catch (ExecutionException e) {
+ if (failure == null) {
+ failure = new
ChannelException(sm.getString("parallelNioSender.send.failed"));
+ }
+ Throwable cause = e.getCause();
+ failure.addFaultyMember(entry.getKey(),
+ cause instanceof Exception ? (Exception) cause : new
IOException(cause));
+ } catch (TimeoutException e) {
+ entry.getValue().cancel(true);
+ if (failure == null) {
+ failure = new ChannelException(
+
sm.getString("parallelNioSender.operation.timedout",
Long.toString(getTimeout())));
+ }
+ failure.addFaultyMember(entry.getKey(), e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ChannelException(e);
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ private void sendSecure(Member member, byte[] data, boolean waitForAck,
long deadline) throws Exception {
+ if (member.getSecurePort() < 0) {
+ throw new
IOException(sm.getString("parallelNioSender.securePortUnavailable"));
+ }
+ Exception failure = null;
+ int maxRetryAttempts = Math.max(0, getMaxRetryAttempts());
+ for (int attempt = 0; attempt <= maxRetryAttempts; attempt++) {
+ try {
+ int timeout = remainingTimeout(deadline);
+ try (SocketChannel socket = SocketChannel.open()) {
+ socket.configureBlocking(true);
+ socket.socket().connect(new
java.net.InetSocketAddress(InetAddress.getByAddress(member.getHost()),
+ member.getSecurePort()), timeout);
+ socket.socket().setSoTimeout(remainingTimeout(deadline));
+ try (TlsChannel tls = new TlsChannel(socket.socket(),
getSslContext().createClientEngine())) {
+ tls.write(java.nio.ByteBuffer.wrap(data));
+ if (waitForAck) {
+
socket.socket().setSoTimeout(remainingTimeout(deadline));
+ readSecureAck(tls, deadline);
+ }
+ }
+ }
+ return;
+ } catch (Exception e) {
+ failure = e;
+ if (System.nanoTime() >= deadline) {
+ break;
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ private void readSecureAck(TlsChannel tls, long deadline) throws
IOException {
+ XByteBuffer acknowledgements = new XByteBuffer(getRxBufSize(), true);
+ java.nio.ByteBuffer buffer =
java.nio.ByteBuffer.allocate(getRxBufSize());
+ while (!acknowledgements.doesPackageExist()) {
+ if (System.nanoTime() >= deadline) {
+ throw new IOException(
+ sm.getString("parallelNioSender.operation.timedout",
Long.toString(getTimeout())));
+ }
+ int read = tls.read(buffer);
+ if (read < 0) {
+ throw new
IOException(sm.getString("nioSender.unable.receive.ack"));
+ }
+ buffer.flip();
+ acknowledgements.append(buffer, read);
+ buffer.clear();
+ }
+ byte[] ack = acknowledgements.extractDataPackage(true).getBytes();
+ if (java.util.Arrays.equals(ack,
org.apache.catalina.tribes.transport.Constants.ACK_DATA)) {
+ return;
+ }
+ if (java.util.Arrays.equals(ack,
org.apache.catalina.tribes.transport.Constants.FAIL_ACK_DATA)) {
+ if (getThrowOnFailedAck()) {
+ throw new
RemoteProcessException(sm.getString("nioSender.receive.failedAck"));
+ }
+ return;
+ }
+ throw new IOException(sm.getString("parallelNioSender.invalidAck"));
+ }
+
+ private int remainingTimeout(long deadline) throws IOException {
+ long remaining = deadline - System.nanoTime();
+ if (remaining <= 0) {
+ throw new
IOException(sm.getString("parallelNioSender.operation.timedout",
Long.toString(getTimeout())));
+ }
+ long milliseconds = TimeUnit.NANOSECONDS.toMillis(remaining);
+ return (int) Math.min(Integer.MAX_VALUE, Math.max(1, milliseconds));
+ }
+
private SendResult doLoop(long selectTimeOut, int maxAttempts, boolean
waitForAck, ChannelMessage msg)
throws ChannelException {
SendResult result = new SendResult();
@@ -383,6 +526,19 @@ public class ParallelNioSender extends AbstractSender
implements MultiPointSende
} catch (Exception ignore) {
// Ignore
}
+ if (sendSecureExecutor != null) {
+ sendSecureExecutor.shutdown();
+ try {
+ // Should stop a lot faster than this as all the sockets have
been closed.
+ if (!sendSecureExecutor.awaitTermination(60,
TimeUnit.SECONDS)) {
+
log.warn(sm.getString("parallelNioSender.disconnect.executor.timeout"));
+ }
+ } catch (InterruptedException e) {
+
log.warn(sm.getString("parallelNioSender.disconnect.executor.interrupted"));
+ } finally {
+ sendSecureExecutor = null;
+ }
+ }
}
@Override
diff --git a/java/org/apache/catalina/tribes/transport/nio/TlsChannel.java
b/java/org/apache/catalina/tribes/transport/nio/TlsChannel.java
new file mode 100644
index 0000000000..6db5243194
--- /dev/null
+++ b/java/org/apache/catalina/tribes/transport/nio/TlsChannel.java
@@ -0,0 +1,222 @@
+/*
+ * 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.catalina.tribes.transport.nio;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.ByteChannel;
+
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLEngineResult.HandshakeStatus;
+import javax.net.ssl.SSLEngineResult.Status;
+
+final class TlsChannel implements ByteChannel {
+
+ private static final ByteBuffer EMPTY = ByteBuffer.allocate(0);
+
+ private final Socket socket;
+ private final InputStream input;
+ private final OutputStream output;
+ private final SSLEngine engine;
+ private ByteBuffer networkInput;
+ private ByteBuffer networkOutput;
+ private ByteBuffer applicationInput;
+
+ TlsChannel(Socket socket, SSLEngine engine) throws IOException {
+ this.socket = socket;
+ input = socket.getInputStream();
+ output = socket.getOutputStream();
+ this.engine = engine;
+ int packetSize = engine.getSession().getPacketBufferSize();
+ networkInput = ByteBuffer.allocate(packetSize);
+ networkOutput = ByteBuffer.allocate(packetSize);
+ applicationInput =
ByteBuffer.allocate(engine.getSession().getApplicationBufferSize());
+ applicationInput.flip();
+ engine.beginHandshake();
+ handshake();
+ }
+
+ private void handshake() throws IOException {
+ HandshakeStatus status = engine.getHandshakeStatus();
+ while (status != HandshakeStatus.FINISHED && status !=
HandshakeStatus.NOT_HANDSHAKING) {
+ switch (status) {
+ case NEED_TASK -> {
+ Runnable task;
+ while ((task = engine.getDelegatedTask()) != null) {
+ task.run();
+ }
+ }
+ case NEED_WRAP -> {
+ networkOutput.clear();
+ SSLEngineResult result = engine.wrap(EMPTY, networkOutput);
+ status = result.getHandshakeStatus();
+ networkOutput.flip();
+ writeFully(networkOutput);
+ continue;
+ }
+ case NEED_UNWRAP, NEED_UNWRAP_AGAIN -> {
+ status = unwrapHandshake();
+ continue;
+ }
+ default -> {
+ // FINISHED and NOT_HANDSHAKING are handled by the loop
condition.
+ }
+ }
+ status = engine.getHandshakeStatus();
+ }
+ applicationInput.clear();
+ applicationInput.flip();
+ }
+
+ @Override
+ public int read(ByteBuffer destination) throws IOException {
+ if (applicationInput.hasRemaining()) {
+ return transfer(applicationInput, destination);
+ }
+ applicationInput.clear();
+ while (true) {
+ if (networkInput.position() == 0 && readEncrypted() < 0) {
+ return -1;
+ }
+ networkInput.flip();
+ SSLEngineResult result = engine.unwrap(networkInput,
applicationInput);
+ networkInput.compact();
+ if (result.getStatus() == Status.BUFFER_OVERFLOW) {
+ applicationInput = expand(applicationInput,
engine.getSession().getApplicationBufferSize());
+ continue;
+ }
+ if (result.getStatus() == Status.BUFFER_UNDERFLOW) {
+ if (applicationInput.position() > 0) {
+ applicationInput.flip();
+ return transfer(applicationInput, destination);
+ }
+ if (!networkInput.hasRemaining()) {
+ networkInput = expand(networkInput,
engine.getSession().getPacketBufferSize());
+ }
+ if (readEncrypted() < 0) {
+ return -1;
+ }
+ continue;
+ }
+ applicationInput.flip();
+ if (result.getStatus() == Status.CLOSED) {
+ return -1;
+ }
+ if (applicationInput.hasRemaining()) {
+ return transfer(applicationInput, destination);
+ }
+ }
+ }
+
+ private HandshakeStatus unwrapHandshake() throws IOException {
+ while (true) {
+ HandshakeStatus handshakeStatus = engine.getHandshakeStatus();
+ if (networkInput.position() == 0 && handshakeStatus !=
HandshakeStatus.NEED_UNWRAP_AGAIN &&
+ readEncrypted() < 0) {
+ throw new EOFException();
+ }
+ networkInput.flip();
+ applicationInput.clear();
+ SSLEngineResult result = engine.unwrap(networkInput,
applicationInput);
+ networkInput.compact();
+ if (result.getStatus() == Status.CLOSED) {
+ throw new EOFException();
+ }
+ if (result.getStatus() == Status.BUFFER_OVERFLOW) {
+ applicationInput = expand(applicationInput,
engine.getSession().getApplicationBufferSize());
+ continue;
+ }
+ if (result.getStatus() == Status.BUFFER_UNDERFLOW) {
+ if (!networkInput.hasRemaining()) {
+ networkInput = expand(networkInput,
engine.getSession().getPacketBufferSize());
+ }
+ if (readEncrypted() < 0) {
+ throw new EOFException();
+ }
+ continue;
+ }
+ return result.getHandshakeStatus();
+ }
+ }
+
+ @Override
+ public int write(ByteBuffer source) throws IOException {
+ int start = source.remaining();
+ while (source.hasRemaining()) {
+ networkOutput.clear();
+ SSLEngineResult result = engine.wrap(source, networkOutput);
+ if (result.getStatus() == Status.CLOSED) {
+ throw new EOFException();
+ }
+ networkOutput.flip();
+ writeFully(networkOutput);
+ }
+ return start;
+ }
+
+ private void writeFully(ByteBuffer buffer) throws IOException {
+ byte[] bytes = new byte[buffer.remaining()];
+ buffer.get(bytes);
+ output.write(bytes);
+ output.flush();
+ }
+
+ private int readEncrypted() throws IOException {
+ byte[] bytes = new byte[networkInput.remaining()];
+ int read = input.read(bytes);
+ if (read > 0) {
+ networkInput.put(bytes, 0, read);
+ }
+ return read;
+ }
+
+ private static ByteBuffer expand(ByteBuffer input, int minimumCapacity) {
+ int capacity = Math.max(input.capacity() * 2, minimumCapacity);
+ ByteBuffer result = ByteBuffer.allocate(capacity);
+ input.flip();
+ result.put(input);
+ return result;
+ }
+
+ private static int transfer(ByteBuffer source, ByteBuffer destination) {
+ int length = Math.min(source.remaining(), destination.remaining());
+ int limit = source.limit();
+ source.limit(source.position() + length);
+ destination.put(source);
+ source.limit(limit);
+ return length;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return !socket.isClosed();
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ engine.closeOutbound();
+ } finally {
+ socket.close();
+ }
+ }
+}
diff --git a/java/org/apache/tomcat/jni/PreSharedKeySelector.java
b/java/org/apache/tomcat/jni/PreSharedKeySelector.java
index 1b4f738f78..2364fdfa10 100644
--- a/java/org/apache/tomcat/jni/PreSharedKeySelector.java
+++ b/java/org/apache/tomcat/jni/PreSharedKeySelector.java
@@ -24,7 +24,7 @@ package org.apache.tomcat.jni;
public interface PreSharedKeySelector {
/**
- * Selects the pre-shared key for the provided identity.
+ * Selects the TLSv1.2 pre-shared key on the server side given the
provided client identity.
*
* @param ssl the SSL instance
* @param identity the PSK identity provided by the client
@@ -36,7 +36,7 @@ public interface PreSharedKeySelector {
byte[] select(long ssl, String identity);
/**
- * Selects the TLSv1.3 pre-shared key and digest for the provided identity.
+ * Selects the TLSv1.3 pre-shared key and digest on the server side given
the provided client identity.
* <p>
* The callback is a little more complex for TLSv1.3. The return value is
still the pre-shared key but OpenSSL also
* needs to know which digest to use. Because the OpenSSL API only exposes
a cipher for this, that is what Tomcat
@@ -51,4 +51,16 @@ public interface PreSharedKeySelector {
* random then 16 bytes are recommended for 128-bit ciphers
and 32 bytes for 256-bit ciphers.
*/
byte[] select(long ssl, byte[] identity, int[] cipherSuite);
+
+
+ /**
+ * Selects the TLS1v2 identity and pre-shared key that the client will
present to a server.
+ *
+ * @param ssl the SSL instance
+ * @param identity a single-element array that must be populated with the
PSK identity
+ *
+ * @return the pre-shared key, or {@code null} if no key is available
+ */
+ byte[] selectClient(long ssl, String[] identity);
+
}
diff --git a/java/org/apache/tomcat/jni/SSLContext.java
b/java/org/apache/tomcat/jni/SSLContext.java
index 8dfc2de0b2..0d9a5ae29f 100644
--- a/java/org/apache/tomcat/jni/SSLContext.java
+++ b/java/org/apache/tomcat/jni/SSLContext.java
@@ -220,7 +220,7 @@ public final class SSLContext {
* @param ctx Server or Client context to use.
* @param cert Certificate file.
* @param key Private Key file to use if not in cert.
- * @param password Certificate password. If null and the certificate is
encrypted, loading the certificate will fail.
+ * @param password Certificate password. If null and certificate is
encrypted, loading the certificate will fail.
* @param idx Certificate index SSL_AIDX_RSA or SSL_AIDX_DSA.
*
* @return <code>true</code> if the operation was successful
@@ -511,7 +511,7 @@ public final class SSLContext {
public static native void setCertVerifyCallback(long ctx,
CertificateVerifier verifier);
/**
- * Allow to hook {@link PreSharedKeySelector} into the TLSv1.2 handshake
processing. This will call
+ * Sets the TLSv1.2 server-side pre-shared key callback to a {@link
PreSharedKeySelector} instance. This will call
* {@code SSL_CTX_set_psk_server_callback}.
*
* @param ctx Server context to use.
@@ -520,7 +520,7 @@ public final class SSLContext {
public static native void setPskServerCallback(long ctx,
PreSharedKeySelector selector);
/**
- * Allow to hook {@link PreSharedKeySelector} into the TLSv1.3 handshake
processing. This will call
+ * Sets the TLSv1.3 server-side pre-shared key callback to a {@link
PreSharedKeySelector} instance. This will call
* {@code SSL_CTX_set_psk_find_session_callback}.
*
* @param ctx Server context to use.
@@ -528,6 +528,14 @@ public final class SSLContext {
*/
public static native void setPskFindSessionCallback(long ctx,
PreSharedKeySelector selector);
+ /**
+ * Sets the TLSv1.2 client-side pre-shared key callback.
+ *
+ * @param ctx Client context to use
+ * @param selector pre-shared key selector
+ */
+ public static native void setPskClientCallback(long ctx,
PreSharedKeySelector selector);
+
/**
* Set application layer protocol for application layer protocol
negotiation extension
*
@@ -552,7 +560,7 @@ public final class SSLContext {
/**
* Set CertificateRaw <br>
- * Use a keystore certificate and key to fill the BIO
+ * Use a keystore certificate and key to fill the BIO.
*
* @param ctx Server or Client context to use.
* @param cert Byte array with the certificate in DER encoding.
@@ -566,7 +574,7 @@ public final class SSLContext {
/**
* Add a certificate to the certificate chain. Certs should be added in
order starting with the issuer of the host
* certs and working up the certificate chain to the CA. <br>
- * Use a keystore certificate chain to fill the BIO
+ * Use a keystore certificate chain to fill the BIO.
*
* @param ctx Server or Client context to use.
* @param cert Byte array with the certificate in DER encoding.
diff --git a/java/org/apache/tomcat/util/net/SSLContext.java
b/java/org/apache/tomcat/util/net/SSLContext.java
index a4d5a0eed1..f51287aa80 100644
--- a/java/org/apache/tomcat/util/net/SSLContext.java
+++ b/java/org/apache/tomcat/util/net/SSLContext.java
@@ -62,6 +62,19 @@ public interface SSLContext {
*/
SSLEngine createSSLEngine();
+ /**
+ * Creates a new SSL engine for the requested mode.
+ *
+ * @param clientMode {@code true} for client mode, otherwise server mode
+ *
+ * @return The new SSL engine
+ */
+ default SSLEngine createSSLEngine(boolean clientMode) {
+ SSLEngine result = createSSLEngine();
+ result.setUseClientMode(clientMode);
+ return result;
+ }
+
/**
* Returns the server socket factory.
*
diff --git a/java/org/apache/tomcat/util/net/openssl/OpenSSLContext.java
b/java/org/apache/tomcat/util/net/openssl/OpenSSLContext.java
index 7d65b98948..e2774e1d9c 100644
--- a/java/org/apache/tomcat/util/net/openssl/OpenSSLContext.java
+++ b/java/org/apache/tomcat/util/net/openssl/OpenSSLContext.java
@@ -93,6 +93,7 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
private final SSLHostConfig sslHostConfig;
private final SSLHostConfigCertificate certificate;
private final List<String> negotiableProtocols;
+ private final boolean clientMode;
private OpenSSLSessionContext sessionContext;
private X509TrustManager x509TrustManager;
@@ -110,8 +111,14 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
* @throws SSLException if initialization fails
*/
public OpenSSLContext(SSLHostConfigCertificate certificate, List<String>
negotiableProtocols) throws SSLException {
+ this(certificate, negotiableProtocols, false);
+ }
+
+ public OpenSSLContext(SSLHostConfigCertificate certificate, List<String>
negotiableProtocols, boolean clientMode)
+ throws SSLException {
this.sslHostConfig = certificate.getSSLHostConfig();
this.certificate = certificate;
+ this.clientMode = clientMode;
long aprPool = Pool.create(0);
long cctx = 0;
long ctx = 0;
@@ -172,7 +179,7 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
// Create SSL Context
try {
- ctx = SSLContext.make(aprPool, value, SSL.SSL_MODE_SERVER);
+ ctx = SSLContext.make(aprPool, value, clientMode ?
SSL.SSL_MODE_CLIENT : SSL.SSL_MODE_SERVER);
} catch (Exception e) {
// If the sslEngine is disabled on the AprLifecycleListener
// there will be an Exception here but there is no way to check
@@ -450,9 +457,11 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
PreSharedKeySelector selector = new
OpenSSLPreSharedKeySelector(psks);
for (String protocol : sslHostConfig.getEnabledProtocols()) {
- if (Constants.SSL_PROTO_TLSv1_2.equals(protocol)) {
+ if (Constants.SSL_PROTO_TLSv1_2.equals(protocol) &&
clientMode) {
+ SSLContext.setPskClientCallback(state.ctx, selector);
+ } else if (Constants.SSL_PROTO_TLSv1_2.equals(protocol)) {
SSLContext.setPskServerCallback(state.ctx, selector);
- } else if (Constants.SSL_PROTO_TLSv1_3.equals(protocol)) {
+ } else if (Constants.SSL_PROTO_TLSv1_3.equals(protocol) &&
!clientMode) {
SSLContext.setPskFindSessionCallback(state.ctx,
selector);
}
}
@@ -650,12 +659,20 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
@Override
public SSLEngine createSSLEngine() {
- return new OpenSSLEngine(cleaner, state.ctx, defaultProtocol, false,
sessionContext,
+ return new OpenSSLEngine(cleaner, state.ctx, defaultProtocol,
clientMode, sessionContext,
(negotiableProtocols != null &&
!negotiableProtocols.isEmpty()), initialized,
sslHostConfig.getCertificateVerificationDepth(),
sslHostConfig.getCertificateVerification() ==
CertificateVerification.OPTIONAL_NO_CA);
}
+ @Override
+ public SSLEngine createSSLEngine(boolean clientMode) {
+ if (clientMode != this.clientMode) {
+ throw new IllegalArgumentException();
+ }
+ return createSSLEngine();
+ }
+
@Override
public SSLServerSocketFactory getServerSocketFactory() {
throw new UnsupportedOperationException();
diff --git
a/java/org/apache/tomcat/util/net/openssl/OpenSSLPreSharedKeySelector.java
b/java/org/apache/tomcat/util/net/openssl/OpenSSLPreSharedKeySelector.java
index 33208edbfc..bbc0bd41e6 100644
--- a/java/org/apache/tomcat/util/net/openssl/OpenSSLPreSharedKeySelector.java
+++ b/java/org/apache/tomcat/util/net/openssl/OpenSSLPreSharedKeySelector.java
@@ -48,13 +48,24 @@ public class OpenSSLPreSharedKeySelector implements
PreSharedKeySelector {
}
}
+ @Override
+ public byte[] selectClient(long ssl, String[] identity) {
+ if (identityToKeyMap.isEmpty()) {
+ return null;
+ }
+ SSLHostConfigPreSharedKey psk =
identityToKeyMap.values().iterator().next();
+ identity[0] = psk.getIdentity();
+ // Need to limit keys to 512 bytes for TLS 1.2
+ return truncateToLength(psk.getIdentity(), psk.getKeyInternal(), 512);
+ }
+
@Override
public byte[] select(long ssl, String identity) {
SSLHostConfigPreSharedKey psk = identityToKeyMap.get(identity);
if (psk == null) {
return null;
}
- // Need to limit keys to 48 bytes for TLS 1.3
+ // Need to limit keys to 512 bytes for TLS 1.2
return truncateToLength(identity, psk.getKeyInternal(), 512);
}
diff --git a/java/org/apache/tomcat/util/net/openssl/panama/OpenSSLContext.java
b/java/org/apache/tomcat/util/net/openssl/panama/OpenSSLContext.java
index 33ff2528ef..36c82320d5 100644
--- a/java/org/apache/tomcat/util/net/openssl/panama/OpenSSLContext.java
+++ b/java/org/apache/tomcat/util/net/openssl/panama/OpenSSLContext.java
@@ -73,6 +73,7 @@ import
org.apache.tomcat.util.openssl.SSL_CTX_set_alpn_select_cb$cb;
import org.apache.tomcat.util.openssl.SSL_CTX_set_cert_verify_callback$cb;
import org.apache.tomcat.util.openssl.SSL_CTX_set_tmp_dh_callback$dh;
import org.apache.tomcat.util.openssl.SSL_CTX_set_verify$callback;
+import org.apache.tomcat.util.openssl.SSL_psk_client_cb_func;
import org.apache.tomcat.util.openssl.SSL_psk_find_session_cb_func;
import org.apache.tomcat.util.openssl.SSL_psk_server_cb_func;
import org.apache.tomcat.util.openssl.openssl_h;
@@ -119,6 +120,7 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
private final SSLHostConfig sslHostConfig;
private final SSLHostConfigCertificate certificate;
+ private final boolean clientMode;
private final boolean alpn;
private final int minTlsVersion;
private final int maxTlsVersion;
@@ -155,7 +157,11 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
}
public OpenSSLContext(SSLHostConfigCertificate certificate, List<String>
negotiableProtocols) throws SSLException {
+ this(certificate, negotiableProtocols, false);
+ }
+ public OpenSSLContext(SSLHostConfigCertificate certificate, List<String>
negotiableProtocols, boolean clientMode)
+ throws SSLException {
// Check that OpenSSL was initialized
if (!OpenSSLStatus.isInitialized()) {
try {
@@ -167,6 +173,7 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
this.sslHostConfig = certificate.getSSLHostConfig();
this.certificate = certificate;
+ this.clientMode = clientMode;
contextArena = Arena.ofAuto();
MemorySegment sslCtx = MemorySegment.NULL;
@@ -193,7 +200,7 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
}
// SSL protocol
- sslCtx = SSL_CTX_new(TLS_server_method());
+ sslCtx = SSL_CTX_new(clientMode ? TLS_client_method() :
TLS_server_method());
if (MemorySegment.NULL.equals(sslCtx)) {
throw new SSLException(sm.getString("openssl.errMakeContext",
OpenSSLLibrary.getLastError()));
}
@@ -654,10 +661,13 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
if (!psks.isEmpty()) {
OpenSSLPreSharedKeySelector selector = new
OpenSSLPreSharedKeySelector(psks);
for (String protocol : sslHostConfig.getEnabledProtocols()) {
- if (Constants.SSL_PROTO_TLSv1_2.equals(protocol)) {
+ if (Constants.SSL_PROTO_TLSv1_2.equals(protocol) &&
clientMode) {
+ SSL_CTX_set_psk_client_callback(state.sslCtx,
+ SSL_psk_client_cb_func.allocate(new
PskClientCallback(selector), contextArena));
+ } else if (Constants.SSL_PROTO_TLSv1_2.equals(protocol)) {
SSL_CTX_set_psk_server_callback(state.sslCtx,
SSL_psk_server_cb_func
.allocate(new PskServerCallback(selector),
contextArena));
- } else if (Constants.SSL_PROTO_TLSv1_3.equals(protocol)) {
+ } else if (Constants.SSL_PROTO_TLSv1_3.equals(protocol) &&
!clientMode) {
if (openssl_h_Compatibility.LIBRESSL) {
throw new
SSLException(sm.getString("openssl.pskTls13Unsupported"));
}
@@ -843,6 +853,39 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
}
}
+ private static class PskClientCallback implements
SSL_psk_client_cb_func.Function {
+
+ private final OpenSSLPreSharedKeySelector selector;
+
+ PskClientCallback(OpenSSLPreSharedKeySelector selector) {
+ this.selector = selector;
+ }
+
+ @Override
+ public int apply(MemorySegment ssl, MemorySegment hint, MemorySegment
identity, int maxIdentityLength,
+ MemorySegment psk, int maxPskLength) {
+ try {
+ String[] selectedIdentity = new String[1];
+ byte[] key = selector.selectClient(ssl.address(),
selectedIdentity);
+ if (key == null || selectedIdentity[0] == null) {
+ return 0;
+ }
+ byte[] identityBytes =
selectedIdentity[0].getBytes(StandardCharsets.UTF_8);
+ if (key.length == 0 || key.length > maxPskLength ||
identityBytes.length + 1 > maxIdentityLength) {
+ return 0;
+ }
+ try (var localArena = Arena.ofConfined()) {
+ MemorySegment identitySegment =
identity.reinterpret(identityBytes.length + 1, localArena, null);
+
identitySegment.copyFrom(localArena.allocateFrom(selectedIdentity[0]));
+ psk.reinterpret(key.length, localArena,
null).copyFrom(MemorySegment.ofArray(key));
+ }
+ return key.length;
+ } catch (RuntimeException e) {
+ return 0;
+ }
+ }
+ }
+
private static class PskFindSessionCallback implements
SSL_psk_find_session_cb_func.Function {
private final OpenSSLPreSharedKeySelector selector;
@@ -1479,12 +1522,20 @@ public class OpenSSLContext implements
org.apache.tomcat.util.net.SSLContext {
@Override
public SSLEngine createSSLEngine() {
- return new OpenSSLEngine(cleaner, state.sslCtx, defaultProtocol,
false, sessionContext, alpn, initialized,
+ return new OpenSSLEngine(cleaner, state.sslCtx, defaultProtocol,
clientMode, sessionContext, alpn, initialized,
sslHostConfig.getCertificateVerificationDepth(),
sslHostConfig.getCertificateVerification() ==
CertificateVerification.OPTIONAL_NO_CA, noOcspCheck,
ocspSoftFail, ocspTimeout, ocspVerifyFlags);
}
+ @Override
+ public SSLEngine createSSLEngine(boolean clientMode) {
+ if (clientMode != this.clientMode) {
+ throw new IllegalArgumentException();
+ }
+ return createSSLEngine();
+ }
+
@Override
public SSLServerSocketFactory getServerSocketFactory() {
throw new UnsupportedOperationException();
diff --git a/java/org/apache/tomcat/util/openssl/SSL_psk_client_cb_func.java
b/java/org/apache/tomcat/util/openssl/SSL_psk_client_cb_func.java
new file mode 100644
index 0000000000..8811151696
--- /dev/null
+++ b/java/org/apache/tomcat/util/openssl/SSL_psk_client_cb_func.java
@@ -0,0 +1,44 @@
+/*
+ * 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.
+ */
+
+// Generated by jextract
+
+package org.apache.tomcat.util.openssl;
+
+import java.lang.foreign.Arena;
+import java.lang.foreign.FunctionDescriptor;
+import java.lang.foreign.Linker;
+import java.lang.foreign.MemorySegment;
+import java.lang.invoke.MethodHandle;
+
+public class SSL_psk_client_cb_func {
+
+ public interface Function {
+ int apply(MemorySegment ssl, MemorySegment hint, MemorySegment
identity, int maxIdentityLength,
+ MemorySegment psk, int maxPskLength);
+ }
+
+ private static final FunctionDescriptor $DESC =
FunctionDescriptor.of(openssl_h.C_INT, openssl_h.C_POINTER,
+ openssl_h.C_POINTER, openssl_h.C_POINTER, openssl_h.C_INT,
openssl_h.C_POINTER, openssl_h.C_INT);
+
+ private static final MethodHandle UP$MH =
+ openssl_h.upcallHandle(SSL_psk_client_cb_func.Function.class,
"apply", $DESC);
+
+ public static MemorySegment allocate(SSL_psk_client_cb_func.Function fi,
Arena scope) {
+ return Linker.nativeLinker().upcallStub(UP$MH.bindTo(fi), $DESC,
scope);
+ }
+}
diff --git a/java/org/apache/tomcat/util/openssl/openssl_h.java
b/java/org/apache/tomcat/util/openssl/openssl_h.java
index e5b3d5da11..4533664eed 100644
--- a/java/org/apache/tomcat/util/openssl/openssl_h.java
+++ b/java/org/apache/tomcat/util/openssl/openssl_h.java
@@ -10465,4 +10465,33 @@ public class openssl_h {
throw new AssertionError("should not reach here", ex$);
}
}
+
+ private static class TLS_client_method {
+ public static final FunctionDescriptor DESC =
FunctionDescriptor.of(openssl_h.C_POINTER);
+ public static final MemorySegment ADDR =
openssl_h.findOrThrow("TLS_client_method");
+ public static final MethodHandle HANDLE =
Linker.nativeLinker().downcallHandle(ADDR, DESC);
+ }
+
+ public static MemorySegment TLS_client_method() {
+ try {
+ return (MemorySegment) TLS_client_method.HANDLE.invokeExact();
+ } catch (Throwable ex$) {
+ throw new AssertionError("should not reach here", ex$);
+ }
+ }
+
+ private static class SSL_CTX_set_psk_client_callback {
+ public static final FunctionDescriptor DESC =
+ FunctionDescriptor.ofVoid(openssl_h.C_POINTER,
openssl_h.C_POINTER);
+ public static final MemorySegment ADDR =
openssl_h.findOrThrow("SSL_CTX_set_psk_client_callback");
+ public static final MethodHandle HANDLE =
Linker.nativeLinker().downcallHandle(ADDR, DESC);
+ }
+
+ public static void SSL_CTX_set_psk_client_callback(MemorySegment ctx,
MemorySegment callback) {
+ try {
+ SSL_CTX_set_psk_client_callback.HANDLE.invokeExact(ctx, callback);
+ } catch (Throwable ex$) {
+ throw new AssertionError("should not reach here", ex$);
+ }
+ }
}
diff --git a/res/openssl/openssl-tomcat.conf b/res/openssl/openssl-tomcat.conf
index e9ac0d0fb8..21bd86fff8 100644
--- a/res/openssl/openssl-tomcat.conf
+++ b/res/openssl/openssl-tomcat.conf
@@ -167,6 +167,7 @@
#### Extracted from: /usr/include/openssl/ssl.h
--include-typedef SSL_psk_find_session_cb_func
+--include-typedef SSL_psk_client_cb_func
--include-typedef SSL_psk_server_cb_func
--include-function OPENSSL_init_ssl # header:
/usr/include/openssl/ssl.h
--include-function SSL_CIPHER_get_auth_nid # header:
/usr/include/openssl/ssl.h
@@ -201,6 +202,7 @@
--include-function SSL_CTX_set_default_passwd_cb # header:
/usr/include/openssl/ssl.h
--include-function SSL_CTX_set_default_verify_paths # header:
/usr/include/openssl/ssl.h
--include-function SSL_CTX_set_options # header:
/usr/include/openssl/ssl.h
+--include-function SSL_CTX_set_psk_client_callback # header:
/usr/include/openssl/ssl.h
--include-function SSL_CTX_set_psk_find_session_callback # header:
/usr/include/openssl/ssl.h
--include-function SSL_CTX_set_psk_server_callback # header:
/usr/include/openssl/ssl.h
--include-function SSL_CTX_set_session_id_context # header:
/usr/include/openssl/ssl.h
@@ -251,6 +253,7 @@
--include-function SSL_verify_client_post_handshake # header:
/usr/include/openssl/ssl.h
--include-function SSL_write # header:
/usr/include/openssl/ssl.h
--include-function TLS_server_method # header:
/usr/include/openssl/ssl.h
+--include-function TLS_client_method # header:
/usr/include/openssl/ssl.h
--include-constant SSL_CB_HANDSHAKE_DONE # header:
/usr/include/openssl/ssl.h
--include-constant SSL_CONF_FLAG_CERTIFICATE # header:
/usr/include/openssl/ssl.h
--include-constant SSL_CONF_FLAG_FILE # header:
/usr/include/openssl/ssl.h
diff --git a/test/org/apache/catalina/tribes/group/TestGroupChannelTls.java
b/test/org/apache/catalina/tribes/group/TestGroupChannelTls.java
new file mode 100644
index 0000000000..4dca0b666b
--- /dev/null
+++ b/test/org/apache/catalina/tribes/group/TestGroupChannelTls.java
@@ -0,0 +1,84 @@
+/*
+ * 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.catalina.tribes.group;
+
+import java.io.Serializable;
+import java.net.ServerSocket;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelListener;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.transport.ReceiverBase;
+
+public class TestGroupChannelTls {
+
+ @Test
+ public void testSecureMessage() throws Exception {
+ try (TribesSslContext ignored =
+ new TribesSslContext("tribes-test",
"000102030405060708090a0b0c0d0e0f")) {
+ // Verify that a supported TLS provider is available before
creating the channels.
+ } catch (Exception e) {
+ Assume.assumeNoException(e);
+ }
+
+ GroupChannel sender = createChannel();
+ GroupChannel receiver = createChannel();
+ CountDownLatch received = new CountDownLatch(1);
+ receiver.addChannelListener(new ChannelListener() {
+ @Override
+ public boolean accept(Serializable msg, Member member) {
+ return true;
+ }
+
+ @Override
+ public void messageReceived(Serializable msg, Member member) {
+ if ("test".equals(msg)) {
+ received.countDown();
+ }
+ }
+ });
+
+ try {
+ sender.start(Channel.SND_RX_SEQ | Channel.SND_TX_SEQ);
+ receiver.start(Channel.SND_RX_SEQ | Channel.SND_TX_SEQ);
+ sender.send(new Member[] { receiver.getLocalMember(false) },
"test",
+ Channel.SEND_OPTIONS_SECURE |
Channel.SEND_OPTIONS_USE_ACK);
+ Assert.assertTrue(received.await(5, TimeUnit.SECONDS));
+ } finally {
+ sender.stop(Channel.DEFAULT);
+ receiver.stop(Channel.DEFAULT);
+ }
+ }
+
+ private static GroupChannel createChannel() throws Exception {
+ GroupChannel channel = new GroupChannel();
+ channel.setPskIdentity("tribes-test");
+ channel.setPskKey("000102030405060708090a0b0c0d0e0f");
+ ReceiverBase receiver = (ReceiverBase) channel.getChannelReceiver();
+ receiver.setHost("localhost");
+ try (ServerSocket socket = new ServerSocket(0)) {
+ receiver.setSecurePort(socket.getLocalPort());
+ }
+ return channel;
+ }
+}
diff --git a/test/org/apache/catalina/tribes/transport/nio/TestTlsChannel.java
b/test/org/apache/catalina/tribes/transport/nio/TestTlsChannel.java
new file mode 100644
index 0000000000..acf1032d8c
--- /dev/null
+++ b/test/org/apache/catalina/tribes/transport/nio/TestTlsChannel.java
@@ -0,0 +1,112 @@
+/*
+ * 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.catalina.tribes.transport.nio;
+
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.FutureTask;
+
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import org.apache.catalina.tribes.group.TribesSslContext;
+
+public class TestTlsChannel {
+
+ @Test
+ public void testPskRoundTrip() throws Exception {
+ try (TribesSslContext context = createContext();
+ ServerSocketChannel server = ServerSocketChannel.open()) {
+ server.bind(new InetSocketAddress("localhost", 0));
+ FutureTask<Void> serverTask = new FutureTask<>(() -> {
+ try (SocketChannel socket = server.accept();
+ TlsChannel tls = new TlsChannel(socket.socket(),
context.createServerEngine())) {
+ Assert.assertEquals("request", read(tls));
+
tls.write(ByteBuffer.wrap("response".getBytes(StandardCharsets.UTF_8)));
+ }
+ return null;
+ });
+ Thread serverThread = new Thread(serverTask);
+ serverThread.start();
+ try (SocketChannel socket =
SocketChannel.open(server.getLocalAddress());
+ TlsChannel tls = new TlsChannel(socket.socket(),
context.createClientEngine())) {
+
tls.write(ByteBuffer.wrap("request".getBytes(StandardCharsets.UTF_8)));
+ Assert.assertEquals("response", read(tls));
+ }
+ serverTask.get();
+ }
+ }
+
+ @Test
+ public void testMultipleRecords() throws Exception {
+ byte[] payload = new byte[128 * 1024];
+ for (int i = 0; i < payload.length; i++) {
+ payload[i] = (byte) i;
+ }
+ try (TribesSslContext context = createContext();
+ ServerSocketChannel server = ServerSocketChannel.open()) {
+ server.bind(new InetSocketAddress("localhost", 0));
+ FutureTask<Void> serverTask = new FutureTask<>(() -> {
+ try (SocketChannel socket = server.accept()) {
+ socket.socket().setSoTimeout(5000);
+ try (TlsChannel tls = new TlsChannel(socket.socket(),
context.createServerEngine())) {
+ Assert.assertArrayEquals(payload, read(tls,
payload.length));
+ tls.write(ByteBuffer.wrap(payload));
+ }
+ }
+ return null;
+ });
+ new Thread(serverTask).start();
+ try (SocketChannel socket =
SocketChannel.open(server.getLocalAddress())) {
+ socket.socket().setSoTimeout(5000);
+ try (TlsChannel tls = new TlsChannel(socket.socket(),
context.createClientEngine())) {
+ tls.write(ByteBuffer.wrap(payload));
+ Assert.assertArrayEquals(payload, read(tls,
payload.length));
+ }
+ }
+ serverTask.get();
+ }
+ }
+
+ private static TribesSslContext createContext() {
+ try {
+ return new TribesSslContext("tribes-test",
"000102030405060708090a0b0c0d0e0f");
+ } catch (Exception e) {
+ Assume.assumeNoException(e);
+ return null;
+ }
+ }
+
+ private static String read(TlsChannel channel) throws Exception {
+ ByteBuffer buffer = ByteBuffer.allocate(32);
+ channel.read(buffer);
+ buffer.flip();
+ return StandardCharsets.UTF_8.decode(buffer).toString();
+ }
+
+ private static byte[] read(TlsChannel channel, int length) throws
Exception {
+ ByteBuffer buffer = ByteBuffer.allocate(length);
+ while (buffer.hasRemaining()) {
+ Assert.assertTrue(channel.read(buffer) > 0);
+ }
+ return buffer.array();
+ }
+}
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index fca666711e..b7bc41b5f6 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -666,6 +666,9 @@
configured address, while the unresolvable host was advertised to the
other cluster members. (remm)
</fix>
+ <add>
+ Add TLS pre-shared key support to Apache Tribes. (markt)
+ </add>
</changelog>
</subsection>
<subsection name="jdbc-pool">
diff --git a/webapps/docs/config/cluster-channel.xml
b/webapps/docs/config/cluster-channel.xml
index aee8271d95..fb682cc211 100644
--- a/webapps/docs/config/cluster-channel.xml
+++ b/webapps/docs/config/cluster-channel.xml
@@ -70,6 +70,11 @@
The default implementation uses non-blocking TCP server sockets.<br/>
You can always find out more about <a
href="../tribes/introduction.html">Apache Tribes</a>
</p>
+
+ <p>TLS pre-shared key support may be configured on the
<strong>Channel</strong>. OpenSSL FFM is preferred when
+ available, followed by Tomcat Native. JSSE is not used for Tribes TLS.
When TLS is enabled, cluster traffic will use
+ blocking socket I/O internally.</p>
+
<p><b><a href="cluster-interceptor.html">Channel/Interceptor</a>:</b> <br/>
The channel will send messages through an interceptor stack. Because of
this, you have the ability to
customize the way messages are sent and received, and even how membership
is handled.<br/>
@@ -117,6 +122,23 @@
flag. The default is false.
</attribute>
+ <attribute name="secure" required="false">
+ If <code>true</code>, the <code>SEND_OPTIONS_SECURE</code> flag is
+ added to every message. Channel startup fails if TLS is unavailable.
+ The default is <code>false</code>.
+ </attribute>
+
+ <attribute name="pskIdentity" required="false">
+ The TLS pre-shared key identity. This must be configured together with
+ <code>pskKey</code> to enable TLS.
+ </attribute>
+
+ <attribute name="pskKey" required="false">
+ The TLS pre-shared key encoded as hexadecimal characters. This must be
+ configured together with <code>pskIdentity</code>. The key is not
+ exposed through JMX. It should be no longer than 512 bytes for TLS 1.2.
+ </attribute>
+
<attribute name="jmxEnabled" required="false">
Flag whether the channel components register with JMX or not.
The default value is true.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
