You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by yw...@apache.org on 2026/08/04 15:35:04 UTC
(systemds) branch main updated: [SYSTEMDS-3955] Support SSL authentication for federated network communication
This is an automated email from the ASF dual-hosted git repository.
ywcb00 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/systemds.git
The following commit(s) were added to refs/heads/main by this push:
new 697db75ff7 [SYSTEMDS-3955] Support SSL authentication for federated network communication
697db75ff7 is described below
commit 697db75ff765b676418482ae9829b861fd8d62b0
Author: Grigorii Turchenko <tu...@gmail.com>
AuthorDate: Tue Aug 4 17:31:34 2026 +0200
[SYSTEMDS-3955] Support SSL authentication for federated network communication
---
.gitignore | 2 +
conf/SystemDS-config.xml.template | 13 ++
pom.xml | 2 +
scripts/tutorials/federated/conf/ssl.xml | 8 +
src/main/java/org/apache/sysds/conf/DMLConfig.java | 7 +
.../controlprogram/federated/FederatedData.java | 31 ++-
.../controlprogram/federated/FederatedSSLUtil.java | 93 ++++++++-
.../controlprogram/federated/FederatedWorker.java | 83 +++-----
.../functions/federated/io/FederatedSSLTest.java | 232 +++++++++++++++------
src/test/resources/cert/ca-cert.pem | 18 ++
src/test/resources/cert/localhost-cert.pem | 37 ++++
src/test/resources/cert/localhost-key.pem | 28 +++
src/test/resources/cert/otherhost-cert.pem | 37 ++++
src/test/resources/cert/otherhost-key.pem | 28 +++
src/test/resources/cert/untrusted-ca-cert.pem | 19 ++
.../federated/io/config/OtherHostSSLConfig.xml | 10 +-
.../federated/io/config/SignedSSLConfig.xml | 10 +-
.../federated/io/config/UntrustedSSLConfig.xml | 8 +-
.../federated/io/generate-certificates.sh | 72 +++++++
19 files changed, 615 insertions(+), 123 deletions(-)
diff --git a/.gitignore b/.gitignore
index acb9435ac6..590413bb8f 100644
--- a/.gitignore
+++ b/.gitignore
@@ -153,6 +153,8 @@ venv/*
# resource optimization
scripts/resource/output
*.pem
+# except the certificates of the federated SSL tests, which are checked in
+!src/test/resources/cert/*.pem
# docker tests
docker/mountFolder/*.bin
diff --git a/conf/SystemDS-config.xml.template b/conf/SystemDS-config.xml.template
index 153dcb6ef2..ced2352b4b 100644
--- a/conf/SystemDS-config.xml.template
+++ b/conf/SystemDS-config.xml.template
@@ -146,6 +146,19 @@
<!-- sets the federated compression strategy (none, zlib, snappy, fastlz, lz4, lzf) -->
<sysds.federated.compression>none</sysds.federated.compression>
+ <!-- enables SSL encrypted federated communication, the certificates below are required with it -->
+ <!-- <sysds.federated.ssl>true</sysds.federated.ssl> -->
+
+ <!-- worker side: the X.509 certificate chain (PEM) and matching PKCS#8 private key (PEM). The certificate has to
+ be issued for the host the coordinator connects to, the coordinator always verifies this. -->
+ <!-- <sysds.federated.ssl.cert>/path/to/worker-cert.pem</sysds.federated.ssl.cert> -->
+ <!-- <sysds.federated.ssl.key>/path/to/worker-key.pem</sysds.federated.ssl.key> -->
+ <!-- if the private key is encrypted, its password is read from the SYSTEMDS_FEDERATED_SSL_KEY_PASSWORD
+ environment variable, do not put it into this file. -->
+
+ <!-- coordinator side: the certificates (PEM) trusted to sign worker certificates, typically the CA certificate. -->
+ <!-- <sysds.federated.ssl.trust>/path/to/ca-cert.pem</sysds.federated.ssl.trust> -->
+
<!-- set buffer pool threshold (max size) in % of total heap -->
<sysds.caching.bufferpoollimit>15</sysds.caching.bufferpoollimit>
diff --git a/pom.xml b/pom.xml
index be50b05a92..2560a38b7b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -734,6 +734,8 @@
<exclude>**/*.mtx</exclude>
<exclude>**/*.mtd</exclude>
<exclude>**/*.out</exclude>
+ <!-- Test certificates, PEM carries no comments -->
+ <exclude>src/test/resources/cert/*.pem</exclude>
<exclude>**/__pycache__/**</exclude>
<exclude>**/part-*</exclude>
<exclude>**/*.keep</exclude>
diff --git a/scripts/tutorials/federated/conf/ssl.xml b/scripts/tutorials/federated/conf/ssl.xml
index f375ba33fe..2a9cedd665 100644
--- a/scripts/tutorials/federated/conf/ssl.xml
+++ b/scripts/tutorials/federated/conf/ssl.xml
@@ -18,4 +18,12 @@
-->
<root>
<sysds.federated.ssl>true</sysds.federated.ssl>
+
+ <!-- Required with SSL: the workers need a certificate issued for the host the coordinator contacts them on
+ and signed by an authority the coordinator trusts. The paths below are placeholders, replace them with
+ your own certificates before using this configuration. An encrypted private key takes its password from
+ the environment variable SYSTEMDS_FEDERATED_SSL_KEY_PASSWORD. -->
+ <sysds.federated.ssl.cert>/path/to/worker-cert.pem</sysds.federated.ssl.cert>
+ <sysds.federated.ssl.key>/path/to/worker-key.pem</sysds.federated.ssl.key>
+ <sysds.federated.ssl.trust>/path/to/ca-cert.pem</sysds.federated.ssl.trust>
</root>
\ No newline at end of file
diff --git a/src/main/java/org/apache/sysds/conf/DMLConfig.java b/src/main/java/org/apache/sysds/conf/DMLConfig.java
index d114ccf69b..3a0829922a 100644
--- a/src/main/java/org/apache/sysds/conf/DMLConfig.java
+++ b/src/main/java/org/apache/sysds/conf/DMLConfig.java
@@ -125,6 +125,9 @@ public class DMLConfig
public static final String EVICTION_SHADOW_BUFFERSIZE = "sysds.gpu.eviction.shadow.bufferSize";
public static final String USE_SSL_FEDERATED_COMMUNICATION = "sysds.federated.ssl"; // boolean
+ public static final String FEDERATED_SSL_CERT = "sysds.federated.ssl.cert"; // Path to the worker X.509 certificate chain in PEM format, required if federated SSL is enabled
+ public static final String FEDERATED_SSL_KEY = "sysds.federated.ssl.key"; // Path to the worker private key in PKCS#8 PEM format, required if federated SSL is enabled
+ public static final String FEDERATED_SSL_TRUST = "sysds.federated.ssl.trust"; // Path to the trusted (CA) certificates in PEM format, required if federated SSL is enabled
public static final String DEFAULT_FEDERATED_INITIALIZATION_TIMEOUT = "sysds.federated.initialization.timeout"; // int seconds
public static final String FEDERATED_TIMEOUT = "sysds.federated.timeout"; // single request timeout default -1 to indicate infinite.
public static final String FEDERATED_PLANNER = "sysds.federated.planner";
@@ -211,6 +214,9 @@ public class DMLConfig
_defaultVals.put(GPU_RULE_BASED_PLACEMENT, "false");
_defaultVals.put(FLOATING_POINT_PRECISION, "double" );
_defaultVals.put(USE_SSL_FEDERATED_COMMUNICATION, "false");
+ _defaultVals.put(FEDERATED_SSL_CERT, null);
+ _defaultVals.put(FEDERATED_SSL_KEY, null);
+ _defaultVals.put(FEDERATED_SSL_TRUST, null);
_defaultVals.put(DEFAULT_FEDERATED_INITIALIZATION_TIMEOUT, "10");
_defaultVals.put(FEDERATED_TIMEOUT, "86400"); // default 1 day compute timeout.
_defaultVals.put(FEDERATED_PLANNER, FederatedPlanner.RUNTIME.name());
@@ -475,6 +481,7 @@ public class DMLConfig
PRINT_GPU_MEMORY_INFO, AVAILABLE_GPUS, SYNCHRONIZE_GPU, EAGER_CUDA_FREE, GPU_RULE_BASED_PLACEMENT,
FLOATING_POINT_PRECISION, GPU_EVICTION_POLICY, LOCAL_SPARK_NUM_THREADS, EVICTION_SHADOW_BUFFERSIZE,
GPU_MEMORY_ALLOCATOR, GPU_MEMORY_UTILIZATION_FACTOR, USE_SSL_FEDERATED_COMMUNICATION,
+ FEDERATED_SSL_CERT, FEDERATED_SSL_KEY, FEDERATED_SSL_TRUST,
DEFAULT_FEDERATED_INITIALIZATION_TIMEOUT, FEDERATED_TIMEOUT, FEDERATED_MONITOR_FREQUENCY, FEDERATED_COMPRESSION,
ASYNC_PREFETCH, ASYNC_SPARK_BROADCAST, ASYNC_SPARK_CHECKPOINT, IO_COMPRESSION_CODEC
};
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java
index 19277ba084..3c6e64ada5 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java
@@ -242,6 +242,8 @@ public class FederatedData {
DataRequestHandler handler) {
final int timeout = ConfigurationManager.getFederatedTimeout();
final boolean ssl = ConfigurationManager.isFederatedSSL();
+ if(ssl)
+ FederatedSSLUtil.SslConstructor();
return new ChannelInitializer<>() {
@Override
@@ -308,18 +310,43 @@ public class FederatedData {
}
private static class DataRequestHandler extends ChannelInboundHandlerAdapter {
+ // The promise is assigned by the requesting thread, while the channel events below are handled on the
+ // event loop, and the two orders are not guaranteed: a rejected SSL handshake already fails the channel
+ // while the requesting thread is still connecting.
+ private final Object _promLock = new Object();
private Promise<FederatedResponse> _prom;
+ // A failure observed before the promise was assigned, e.g., a rejected SSL handshake
+ private Throwable _failure;
public DataRequestHandler() {
}
public void setPromise(Promise<FederatedResponse> prom) {
- _prom = prom;
+ synchronized(_promLock) {
+ _prom = prom;
+ if(_failure != null)
+ _prom.tryFailure(_failure);
+ }
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
- _prom.setSuccess((FederatedResponse) msg);
+ synchronized(_promLock) {
+ _prom.setSuccess((FederatedResponse) msg);
+ }
+ ctx.close();
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ // fail the request instead of waiting for a response that never arrives
+ // covers a failed SSL handshake, e.g., if the worker certificate is not signed by a trusted authority.
+ synchronized(_promLock) {
+ if(_prom != null)
+ _prom.tryFailure(cause);
+ else
+ _failure = cause;
+ }
ctx.close();
}
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedSSLUtil.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedSSLUtil.java
index f1300ef0f4..33a89a5a4f 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedSSLUtil.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedSSLUtil.java
@@ -19,19 +19,29 @@
package org.apache.sysds.runtime.controlprogram.federated;
+import java.io.File;
import java.net.InetSocketAddress;
+import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLParameters;
+import org.apache.log4j.Logger;
+import org.apache.sysds.conf.ConfigurationManager;
+import org.apache.sysds.conf.DMLConfig;
import org.apache.sysds.runtime.DMLRuntimeException;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslHandler;
-import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
public class FederatedSSLUtil {
+ private static final Logger LOG = Logger.getLogger(FederatedSSLUtil.class);
+
+ // The password of an encrypted worker private key is read from the environment instead of the configuration,
+ // so that it is not leaked when the configuration file is shared or published.
+ public static final String SSL_KEY_PASSWORD_ENV = "SYSTEMDS_FEDERATED_SSL_KEY_PASSWORD";
private FederatedSSLUtil(){
// private constructor.
@@ -40,24 +50,93 @@ public class FederatedSSLUtil {
/** A Singleton constructed SSL context, that only is assigned if ssl is enabled. */
private static SslContextMan sslInstance = null;
- protected static SslContextMan SslConstructor() {
+ protected synchronized static SslContextMan SslConstructor() {
if(sslInstance == null)
- return new SslContextMan();
- else
- return sslInstance;
+ sslInstance = new SslContextMan();
+ return sslInstance;
+ }
+
+ // Drop the cached client side SSL context, so that the next connection is built from the current configuration.
+ // Only relevant if the configuration changes while the JVM is running, as it does in tests.
+ public synchronized static void resetClientContext() {
+ sslInstance = null;
}
protected static SslHandler createSSLHandler(SocketChannel ch, InetSocketAddress address) {
- return SslConstructor().context.newHandler(ch.alloc(), address.getAddress().getHostAddress(), address.getPort());
+ final SslContextMan man = SslConstructor();
+ // prefer the configured host name over the resolved address, since certificates are issued for host names.
+ final String host = (address.getHostString() != null) ? address.getHostString() : address.getAddress()
+ .getHostAddress();
+ final SslHandler handler = man.context.newHandler(ch.alloc(), host, address.getPort());
+
+ // the certificate of a worker has to be issued for the host it is contacted on, otherwise any worker
+ // with a trusted certificate could impersonate any other worker.
+ final SSLEngine engine = handler.engine();
+ final SSLParameters params = engine.getSSLParameters();
+ params.setEndpointIdentificationAlgorithm("HTTPS");
+ engine.setSSLParameters(params);
+
+ return handler;
+ }
+
+ /**
+ * Construct the SSL context of a federated worker, based on the certificate and private key configured via
+ * {@link DMLConfig#FEDERATED_SSL_CERT} and {@link DMLConfig#FEDERATED_SSL_KEY}. Both are required, a worker that
+ * cannot be authenticated by the coordinator is not supported. If the private key is encrypted, its password is
+ * read from the {@link #SSL_KEY_PASSWORD_ENV} environment variable.
+ *
+ * @return The server side SSL context of the federated worker
+ */
+ public static SslContext createServerContext() {
+ final DMLConfig conf = ConfigurationManager.getDMLConfig();
+ final String certPath = conf.getTextValue(DMLConfig.FEDERATED_SSL_CERT);
+ final String keyPath = conf.getTextValue(DMLConfig.FEDERATED_SSL_KEY);
+ final String keyPassword = System.getenv(SSL_KEY_PASSWORD_ENV);
+
+ if(!isSet(certPath) || !isSet(keyPath))
+ throw new DMLRuntimeException("Federated SSL requires a signed certificate, configure the certificate "
+ + "chain in " + DMLConfig.FEDERATED_SSL_CERT + " and the matching private key in "
+ + DMLConfig.FEDERATED_SSL_KEY + ".");
+
+ try {
+ LOG.info("Federated worker SSL using certificate: " + certPath);
+ return SslContextBuilder
+ .forServer(readableFile(certPath, DMLConfig.FEDERATED_SSL_CERT),
+ readableFile(keyPath, DMLConfig.FEDERATED_SSL_KEY), isSet(keyPassword) ? keyPassword : null)
+ .build();
+ }
+ catch(SSLException e) {
+ throw new DMLRuntimeException("Static SSL setup failed for worker side", e);
+ }
+ }
+
+ private static boolean isSet(String value) {
+ return value != null && !value.trim().isEmpty();
}
+ private static File readableFile(String path, String configName) {
+ final File f = new File(path.trim());
+ if(!f.canRead())
+ throw new DMLRuntimeException(
+ "Federated SSL file configured in " + configName + " is not a readable file: " + path);
+ return f;
+ }
private static class SslContextMan {
protected final SslContext context;
private SslContextMan() {
+ final DMLConfig conf = ConfigurationManager.getDMLConfig();
+ final String trustPath = conf.getTextValue(DMLConfig.FEDERATED_SSL_TRUST);
+
+ if(!isSet(trustPath))
+ throw new DMLRuntimeException("Federated SSL requires the certificates that are trusted to sign "
+ + "worker certificates, configure them in " + DMLConfig.FEDERATED_SSL_TRUST + ".");
+
try {
- context = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
+ LOG.debug("Federated SSL trusting certificates in: " + trustPath);
+ context = SslContextBuilder.forClient()
+ .trustManager(readableFile(trustPath, DMLConfig.FEDERATED_SSL_TRUST)).build();
}
catch(SSLException e) {
throw new DMLRuntimeException("Static SSL setup failed for client side", e);
diff --git a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java
index fc8989053b..682cc8e3ff 100644
--- a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java
+++ b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorker.java
@@ -20,20 +20,16 @@
package org.apache.sysds.runtime.controlprogram.federated;
import java.io.Serializable;
-import java.security.cert.CertificateException;
import java.util.Optional;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import javax.net.ssl.SSLException;
-
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.log4j.Logger;
import org.apache.sysds.api.DMLScript;
import org.apache.sysds.conf.ConfigurationManager;
import org.apache.sysds.conf.DMLConfig;
-import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.controlprogram.caching.CacheBlock;
import org.apache.sysds.runtime.controlprogram.federated.compression.CompressionDecoderEndStatisticsHandler;
import org.apache.sysds.runtime.controlprogram.federated.compression.CompressionDecoderStartStatisticsHandler;
@@ -63,8 +59,6 @@ import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.SslContextBuilder;
-import io.netty.handler.ssl.util.SelfSignedCertificate;
import io.netty.util.concurrent.DefaultThreadFactory;
@SuppressWarnings("deprecation")
@@ -121,12 +115,16 @@ public class FederatedWorker {
LOG.info("Started Federated Worker at port: " + _port);
f.channel().closeFuture().sync();
}
- catch(Exception e) {
+ catch(InterruptedException e) {
LOG.info("Federated worker interrupted");
- if(_debug) {
- LOG.error(e.getMessage());
+ if(_debug)
+ e.printStackTrace();
+ }
+ catch(Exception e) {
+ // report why the worker stops, e.g., a missing certificate with ssl enabled, otherwise it exits silently
+ LOG.error("Federated worker stopped: " + e.getMessage());
+ if(_debug)
e.printStackTrace();
- }
}
finally {
LOG.info("Federated Worker Shutting down.");
@@ -192,47 +190,30 @@ public class FederatedWorker {
}
private ChannelInitializer<SocketChannel> createChannel(boolean ssl) {
- try {
- // TODO add ability to use real ssl files, not self signed certificates.
- final SelfSignedCertificate cert;
- final SslContext cont2;
- final boolean sslEnabled = ConfigurationManager.getDMLConfig().getBooleanValue(DMLConfig.USE_SSL_FEDERATED_COMMUNICATION) || ssl;
-
- if(ssl) {
- cert = new SelfSignedCertificate();
- cont2 = SslContextBuilder.forServer(cert.certificate(), cert.privateKey()).build();
- }
- else {
- cert = null;
- cont2 = null;
+ final SslContext sslContext = ssl ? FederatedSSLUtil.createServerContext() : null;
+
+ return new ChannelInitializer<>() {
+ @Override
+ public void initChannel(SocketChannel ch) {
+ final ChannelPipeline cp = ch.pipeline();
+ if(sslContext != null)
+ cp.addLast(sslContext.newHandler(ch.alloc()));
+
+ final Optional<ImmutablePair<ChannelInboundHandlerAdapter, ChannelOutboundHandlerAdapter>> compressionStrategy = FederationUtils
+ .compressionStrategy();
+ cp.addLast("NetworkTrafficCounter", new NetworkTrafficCounter(FederatedStatistics::logWorkerTraffic));
+ cp.addLast("CompressionDecodingStartStatistics", new CompressionDecoderStartStatisticsHandler());
+ compressionStrategy.ifPresent(strategy -> cp.addLast("CompressionDecoder", strategy.left));
+ cp.addLast("CompressionDecoderEndStatistics", new CompressionDecoderEndStatisticsHandler());
+ cp.addLast("ObjectDecoder", new ObjectDecoder(Integer.MAX_VALUE,
+ ClassResolvers.weakCachingResolver(ClassLoader.getSystemClassLoader())));
+ cp.addLast("CompressionEncodingEndStatistics", new CompressionEncoderEndStatisticsHandler());
+ compressionStrategy.ifPresent(strategy -> cp.addLast("CompressionEncoder", strategy.right));
+ cp.addLast("CompressionEncodingStartStatistics", new CompressionEncoderStartStatisticsHandler());
+ cp.addLast("ObjectEncoder", new ObjectEncoder());
+ cp.addLast(FederationUtils.decoder(), new FederatedResponseEncoder());
+ cp.addLast(new FederatedWorkerHandler(_flt, _frc, _fan, networkTimer));
}
-
- return new ChannelInitializer<>() {
- @Override
- public void initChannel(SocketChannel ch) {
- final ChannelPipeline cp = ch.pipeline();
- if(sslEnabled)
- cp.addLast(cont2.newHandler(ch.alloc()));
-
- final Optional<ImmutablePair<ChannelInboundHandlerAdapter, ChannelOutboundHandlerAdapter>> compressionStrategy = FederationUtils.compressionStrategy();
- cp.addLast("NetworkTrafficCounter", new NetworkTrafficCounter(FederatedStatistics::logWorkerTraffic));
- cp.addLast("CompressionDecodingStartStatistics", new CompressionDecoderStartStatisticsHandler());
- compressionStrategy.ifPresent(strategy -> cp.addLast("CompressionDecoder", strategy.left));
- cp.addLast("CompressionDecoderEndStatistics", new CompressionDecoderEndStatisticsHandler());
- cp.addLast("ObjectDecoder",
- new ObjectDecoder(Integer.MAX_VALUE,
- ClassResolvers.weakCachingResolver(ClassLoader.getSystemClassLoader())));
- cp.addLast("CompressionEncodingEndStatistics", new CompressionEncoderEndStatisticsHandler());
- compressionStrategy.ifPresent(strategy -> cp.addLast("CompressionEncoder", strategy.right));
- cp.addLast("CompressionEncodingStartStatistics", new CompressionEncoderStartStatisticsHandler());
- cp.addLast("ObjectEncoder", new ObjectEncoder());
- cp.addLast(FederationUtils.decoder(), new FederatedResponseEncoder());
- cp.addLast(new FederatedWorkerHandler(_flt, _frc, _fan, networkTimer));
- }
- };
- }
- catch(CertificateException | SSLException e) {
- throw new DMLRuntimeException("Failed creating channel SSL", e);
- }
+ };
}
}
diff --git a/src/test/java/org/apache/sysds/test/functions/federated/io/FederatedSSLTest.java b/src/test/java/org/apache/sysds/test/functions/federated/io/FederatedSSLTest.java
index 5f5c09e07c..ac462e447a 100644
--- a/src/test/java/org/apache/sysds/test/functions/federated/io/FederatedSSLTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/federated/io/FederatedSSLTest.java
@@ -18,132 +18,244 @@
*/
package org.apache.sysds.test.functions.federated.io;
-
import java.io.File;
-import java.util.Arrays;
-import java.util.Collection;
+import java.net.InetSocketAddress;
+import java.security.cert.CertificateException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import javax.net.ssl.SSLException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.sysds.common.Types;
+import org.apache.sysds.conf.ConfigurationManager;
+import org.apache.sysds.conf.DMLConfig;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysds.runtime.controlprogram.federated.FederatedData;
+import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest;
+import org.apache.sysds.runtime.controlprogram.federated.FederatedRequest.RequestType;
+import org.apache.sysds.runtime.controlprogram.federated.FederatedResponse;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.controlprogram.federated.FederatedSSLUtil;
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
import org.apache.sysds.test.AutomatedTestBase;
import org.apache.sysds.test.TestConfiguration;
import org.apache.sysds.test.TestUtils;
import org.apache.sysds.test.functions.federated.FederatedTestObjectConstructor;
import org.junit.Assert;
-import org.junit.Ignore;
import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-@RunWith(value = Parameterized.class)
@net.jcip.annotations.NotThreadSafe
public class FederatedSSLTest extends AutomatedTestBase {
private static final Log LOG = LogFactory.getLog(FederatedSSLTest.class.getName());
- // This test use the same scripts as the Federated Reader tests, just with SSL enabled.
+ // These tests use the same scripts as the Federated Reader tests, just with SSL enabled.
private final static String TEST_DIR = "functions/federated/io/";
private final static String TEST_NAME = "FederatedReaderTest";
private final static String TEST_CLASS_DIR = TEST_DIR + FederatedSSLTest.class.getSimpleName() + "/";
private final static int blocksize = 1024;
- private final static File TEST_CONF_FILE = new File(SCRIPT_DIR + TEST_DIR + "SSLConfig.xml");
+ private final static int rows = 10;
+ private final static int cols = 13;
- @Parameterized.Parameter()
- public int rows;
- @Parameterized.Parameter(1)
- public int cols;
- @Parameterized.Parameter(2)
- public boolean rowPartitioned;
- @Parameterized.Parameter(3)
- public int fedCount;
+ private final static String CONF_DIR = SCRIPT_DIR + TEST_DIR + "config/";
+ // Certificate issued for localhost and signed by the authority the coordinator trusts
+ private final static File SIGNED_CONF = new File(CONF_DIR, "SignedSSLConfig.xml");
+ // Coordinator trusting an authority unrelated to the one that signed the worker certificate
+ private final static File UNTRUSTED_CONF = new File(CONF_DIR, "UntrustedSSLConfig.xml");
+ // Certificate signed by the trusted authority, but issued for another host
+ private final static File OTHER_HOST_CONF = new File(CONF_DIR, "OtherHostSSLConfig.xml");
+ // SSL enabled without configuring any certificate, which is not supported
+ private final static File NO_CERT_CONF = new File(SCRIPT_DIR + TEST_DIR + "SSLConfig.xml");
+
+ private File confFile = SIGNED_CONF;
+ private File workerConfFile = null;
@Override
public void setUp() {
TestUtils.clearAssertionInformation();
addTestConfiguration(TEST_NAME, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME));
+ // The SSL context of the coordinator is cached for the JVM, so it should be cleared in between the tests
+ FederatedSSLUtil.resetClientContext();
+ FederatedData.resetFederatedSites();
}
- @Parameterized.Parameters
- public static Collection<Object[]> data() {
- // number of rows or cols has to be >= number of federated locations.
- return Arrays.asList(new Object[][] {{10, 13, true, 2}});
+ // A certificate signed by the trusted authority and issued for the contacted host is accepted.
+ @Test
+ public void federatedSinglenodeReadSigned() {
+ // the workers and the coordinator share one configuration, the coordinator trusts the signing authority
+ confFile = SIGNED_CONF;
+ federatedRead(Types.ExecMode.SINGLE_NODE);
}
+ // A certificate signed by an authority the coordinator does not know is rejected.
@Test
- @Ignore
- public void federatedSinglenodeRead() {
- federatedRead(Types.ExecMode.SINGLE_NODE);
+ public void untrustedCertificateIsRejected() {
+ workerConfFile = SIGNED_CONF;
+ confFile = UNTRUSTED_CONF;
+ assertRequestFails();
+ }
+
+ // A certificate signed by the trusted authority, but issued for another host, is rejected as well.
+ @Test
+ public void certificateOfOtherHostIsRejected() {
+ confFile = OTHER_HOST_CONF;
+ assertRequestFails();
}
- public void federatedRead(Types.ExecMode execMode) {
- Types.ExecMode oldPlatform = setExecMode(execMode);
+ // A worker cannot serve SSL without a certificate, generating one is not supported.
+ @Test
+ public void workerCertificateIsRequired() throws Exception {
+ confFile = NO_CERT_CONF;
+ getAndLoadTestConfiguration(TEST_NAME);
+ ConfigurationManager.setGlobalConfig(new DMLConfig(getCurConfigFile().getPath()));
+
+ try {
+ FederatedSSLUtil.createServerContext();
+ Assert.fail("A worker without a configured certificate should not start SSL.");
+ }
+ catch(DMLRuntimeException e) {
+ Assert.assertTrue("Expected a hint at the certificate configuration but got: " + e.getMessage(),
+ e.getMessage().contains(DMLConfig.FEDERATED_SSL_CERT));
+ }
+ }
+
+ // A coordinator cannot connect without knowing which certificates to trust.
+ @Test
+ public void trustedCertificatesAreRequired() {
+ workerConfFile = SIGNED_CONF;
+ confFile = NO_CERT_CONF;
+ getAndLoadTestConfiguration(TEST_NAME);
+ fullDMLScriptName = "";
+ final int port = getRandomAvailablePort();
+ final Thread[] workers = startWorkers(port);
+
+ try {
+ // the worker startup sets the global configuration in this JVM, therefore the coordinator
+ // configuration has to be applied after the workers are up.
+ ConfigurationManager.setGlobalConfig(new DMLConfig(getCurConfigFile().getPath()));
+
+ FederatedData.executeFederatedOperation(new InetSocketAddress("localhost", port),
+ new FederatedRequest(RequestType.CLEAR)).get(FED_WORKER_WAIT, TimeUnit.MILLISECONDS);
+ Assert.fail("A coordinator without trusted certificates should not connect.");
+ }
+ catch(Exception e) {
+ Assert.assertTrue("Expected a hint at the trust configuration but got: " + e,
+ messageContains(e, DMLConfig.FEDERATED_SSL_TRUST));
+ }
+ finally {
+ TestUtils.shutdownThreads(workers);
+ }
+ }
+
+ // Read a federated matrix over SSL and compare against the same matrix read locally.
+ private void federatedRead(Types.ExecMode execMode) {
+ final Types.ExecMode oldPlatform = setExecMode(execMode);
getAndLoadTestConfiguration(TEST_NAME);
setOutputBuffering(true);
-
+
// write input matrices
- int halfRows = rows / 2;
- long[][] begins = new long[][] {new long[] {0, 0}, new long[] {halfRows, 0}};
- long[][] ends = new long[][] {new long[] {halfRows, cols}, new long[] {rows, cols}};
+ final int halfRows = rows / 2;
+ final long[][] begins = new long[][] {new long[] {0, 0}, new long[] {halfRows, 0}};
+ final long[][] ends = new long[][] {new long[] {halfRows, cols}, new long[] {rows, cols}};
// We have two matrices handled by a single federated worker
- double[][] X1 = getRandomMatrix(halfRows, cols, 0, 1, 1, 42);
- double[][] X2 = getRandomMatrix(halfRows, cols, 0, 1, 1, 1340);
+ final double[][] X1 = getRandomMatrix(halfRows, cols, 0, 1, 1, 42);
+ final double[][] X2 = getRandomMatrix(halfRows, cols, 0, 1, 1, 1340);
writeInputMatrixWithMTD("X1", X1, false, new MatrixCharacteristics(halfRows, cols, blocksize, halfRows * cols));
writeInputMatrixWithMTD("X2", X2, false, new MatrixCharacteristics(halfRows, cols, blocksize, halfRows * cols));
// empty script name because we don't execute any script, just start the worker
fullDMLScriptName = "";
- int port1 = getRandomAvailablePort();
- int port2 = getRandomAvailablePort();
- Thread[] workers = startLocalFedWorkerThreads(new int[] {port1, port2}, null, FED_WORKER_WAIT);
- String host = "localhost";
+ final int port1 = getRandomAvailablePort();
+ final int port2 = getRandomAvailablePort();
+ final Thread[] workers = startWorkers(port1, port2);
-
try {
- MatrixObject fed = FederatedTestObjectConstructor.constructFederatedInput(
- rows, cols, blocksize, host, begins, ends, new int[] {port1, port2},
- new String[] {input("X1"), input("X2")}, input("X.json"));
- //FIXME: reset avoids deadlock on reference script
- //(because federated matrix creation added to federated sites - blocks on clear)
- //However, there seems to be a regression regarding the SSL handling in general
+ final MatrixObject fed = FederatedTestObjectConstructor.constructFederatedInput(rows, cols, blocksize,
+ "localhost", begins, ends, new int[] {port1, port2}, new String[] {input("X1"), input("X2")},
+ input("X.json"));
+ // FIXME: reset avoids deadlock on reference script
+ // (because federated matrix creation added to federated sites - blocks on clear)
FederatedData.resetFederatedSites();
writeInputFederatedWithMTD("X.json", fed);
+
// Run reference dml script with normal matrix
- fullDMLScriptName = SCRIPT_DIR + "functions/federated/io/" + TEST_NAME + (rowPartitioned ? "Row" : "Col")
- + "2Reference.dml";
+ fullDMLScriptName = SCRIPT_DIR + TEST_DIR + TEST_NAME + "Row2Reference.dml";
programArgs = new String[] {"-stats", "-args", input("X1"), input("X2")};
- String refOut = runTest(null).toString();
-
+ final String refOut = runTest(null).toString();
+
// Run federated
- fullDMLScriptName = SCRIPT_DIR + "functions/federated/io/" + TEST_NAME + ".dml";
+ fullDMLScriptName = SCRIPT_DIR + TEST_DIR + TEST_NAME + ".dml";
programArgs = new String[] {"-stats", "-args", input("X.json")};
- String out = runTest(null).toString();
+ final String out = runTest(null).toString();
Assert.assertTrue(heavyHittersContainsString("fed_uak+"));
// Verify output
- Assert.assertEquals(Double.parseDouble(refOut.split("\n")[0]),
- Double.parseDouble(out.split("\n")[0]), 0.00001);
+ Assert.assertEquals(Double.parseDouble(refOut.split("\n")[0]), Double.parseDouble(out.split("\n")[0]),
+ 0.00001);
}
catch(Exception e) {
e.printStackTrace();
- Assert.assertTrue(false);
+ Assert.fail("Federated read over SSL failed: " + e.getMessage());
}
finally {
resetExecMode(oldPlatform);
+ TestUtils.shutdownThreads(workers);
+ }
+ }
+
+ // The coordinator rejects the certificate of the worker instead of getting an answer to its request.
+ private void assertRequestFails() {
+ getAndLoadTestConfiguration(TEST_NAME);
+ fullDMLScriptName = "";
+ final int port = getRandomAvailablePort();
+ final Thread[] workers = startWorkers(port);
+
+ try {
+ // the worker startup sets the global configuration in this JVM, therefore the coordinator
+ // configuration has to be applied after the workers are up.
+ ConfigurationManager.setGlobalConfig(new DMLConfig(getCurConfigFile().getPath()));
+
+ final Future<FederatedResponse> f = FederatedData.executeFederatedOperation(
+ new InetSocketAddress("localhost", port), new FederatedRequest(RequestType.CLEAR));
+
+ f.get(FED_WORKER_WAIT, TimeUnit.MILLISECONDS);
+ Assert.fail("The request to a worker with a rejected certificate should not succeed.");
}
+ catch(ExecutionException e) {
+ Assert.assertTrue("Expected an SSL failure but got: " + e.getCause(), isSSLFailure(e.getCause()));
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ Assert.fail("Expected an SSL failure but got: " + e);
+ }
+ finally {
+ TestUtils.shutdownThreads(workers);
+ }
+ }
+
+ // The workers have to be started with a configuration as well, otherwise they do not enable SSL.
+ private Thread[] startWorkers(int... ports) {
+ final File conf = (workerConfFile != null) ? workerConfFile : getCurConfigFile();
+ return startLocalFedWorkerThreads(ports, new String[] {"-config", conf.getPath()}, FED_WORKER_WAIT);
+ }
+
+ private static boolean isSSLFailure(Throwable t) {
+ for(Throwable c = t; c != null; c = c.getCause())
+ if(c instanceof SSLException || c instanceof CertificateException)
+ return true;
+ return false;
+ }
- TestUtils.shutdownThreads(workers);
+ private static boolean messageContains(Throwable t, String text) {
+ for(Throwable c = t; c != null; c = c.getCause())
+ if(c.getMessage() != null && c.getMessage().contains(text))
+ return true;
+ return false;
}
- /**
- * Override default configuration with custom test configuration to ensure
- * scratch space and local temporary directory locations are also updated.
- */
@Override
protected File getConfigTemplateFile() {
- // Instrumentation in this test's output log to show custom configuration file used for template.
- LOG.info("This test case overrides default configuration with " + TEST_CONF_FILE.getPath());
- return TEST_CONF_FILE;
+ return confFile;
}
}
diff --git a/src/test/resources/cert/ca-cert.pem b/src/test/resources/cert/ca-cert.pem
new file mode 100644
index 0000000000..5a6e3a18c2
--- /dev/null
+++ b/src/test/resources/cert/ca-cert.pem
@@ -0,0 +1,18 @@
+-----BEGIN CERTIFICATE-----
+MIIC7TCCAdWgAwIBAgIJAIxu3kp0I0VqMA0GCSqGSIb3DQEBCwUAMBsxGTAXBgNV
+BAMTEFN5c3RlbURTIFRlc3QgQ0EwIBcNMjYwNzI5MTIyOTA4WhgPMjEyNjA3MDUx
+MjI5MDhaMBsxGTAXBgNVBAMTEFN5c3RlbURTIFRlc3QgQ0EwggEiMA0GCSqGSIb3
+DQEBAQUAA4IBDwAwggEKAoIBAQCr0ehP5tRAlvTEPNj0nx5D0sJTQRkdMCs3VERt
+dDRebaecIOZse5Uve4r5qKr/YyTnscFZvl4CeM04AKXljhcUPeea/e2eeQegh7IF
+bCPWJXcbTFe4LzJ89asZZEqV7BN0IfDU9OoGKyPILF/1UJ79e8N89KoIiHZYFl9o
+Lj4KlD3dWKRluWG6TWYxBYcnBt2D/cFXXaEa62rqoVp1hIQ/MF232V9eOBvYiCPs
+IYnhPAIjkYz8LhYp02azeULGD3mrck08AFeGjLuXHBVWsFkcM0nb2stOenhkKj8k
+MqbNVA8/iFPrrbkCLROffZnIa0JficXAR2dbe7eUXhHnCpt1AgMBAAGjMjAwMB0G
+A1UdDgQWBBQFgrHcX+3rTCqFWbekJ4qIrH68izAPBgNVHRMBAf8EBTADAQH/MA0G
+CSqGSIb3DQEBCwUAA4IBAQCfHThIA+XEdpP3a/ySBNS2yrsrKe1G6JDRdzMNQiQv
+t1ymTQ1Dm7bDUo7L1fFGX/cMjCo3+2Q5owIGO4t9DYJ6cpSBAad1HLNVmQOTtIeD
+/IXwCpODWe4ZL/zcuJcCbidLuEsy56fLNeuf2fkkWCaGL6ehpHsp09MFwgrU7xLa
+MpPBchHsKmbx8QbpQBYpMSGKvS9uTtrRyEy6OKyyXJfsfR6GlJbVH1cgUN+eg0cp
+PiCbrd6PyeobawteRrfJkCqKvh6/nvdSHXnkfmz47RMC2Obs41LX43Boh5UdHw65
+aJfjh7O8CCHALCWHkvlJOfbEI6s1fj79H98r1//ifY+8
+-----END CERTIFICATE-----
diff --git a/src/test/resources/cert/localhost-cert.pem b/src/test/resources/cert/localhost-cert.pem
new file mode 100644
index 0000000000..d85ca651bf
--- /dev/null
+++ b/src/test/resources/cert/localhost-cert.pem
@@ -0,0 +1,37 @@
+-----BEGIN CERTIFICATE-----
+MIIDETCCAfmgAwIBAgIIKPKZck8b4KYwDQYJKoZIhvcNAQELBQAwGzEZMBcGA1UE
+AxMQU3lzdGVtRFMgVGVzdCBDQTAgFw0yNjA3MjkxMjI5MTBaGA8yMTI2MDcwNTEy
+MjkxMFowFDESMBAGA1UEAxMJbG9jYWxob3N0MIIBIjANBgkqhkiG9w0BAQEFAAOC
+AQ8AMIIBCgKCAQEA5j66iS7szE2psgbLYeLxTlRdSeXhMroUVAhO0jVo59Jafc3S
+3sdrOncf1IbUt5Tq8CCVp2dKmRH0TYnWVFsHj8aJzOz4zfTXjlFSbLMWNf64kLM7
+F0F8KxlInmXnMJNrBYDQe+fVw7tjx7eBhahpfLtti7cQZHF8mWirIi3N4fRcTUQZ
+tLujsLtL/w8Jl6ef/MZ6zo29Rqg8SsgVtZSWz0m/pS8NfvrIY7I/2lbixaV9HlQD
+fCapODmwo5wRSQFHg4PcykZhFUsqV6nOcCmqwfaMlFIhdVRTyvYYO5kDdgqvyOJX
+NYwgHCEckHXiOywwfbZAPzhe5kRDOh8mRNxZrQIDAQABo14wXDAdBgNVHQ4EFgQU
+bYjG7TKhIoelk+qGTt3UtoFiZ5AwGgYDVR0RBBMwEYIJbG9jYWxob3N0hwR/AAAB
+MB8GA1UdIwQYMBaAFAWCsdxf7etMKoVZt6QnioisfryLMA0GCSqGSIb3DQEBCwUA
+A4IBAQBCkX9kwwZSt+N01Pa9QrlRYyqIJJaUTJXz4X8MxhZ8OrmjV87b0lUDf1wQ
+0jlIRIenR/pl1438O/4LEIJb3tErajn8+QqThoDSrlQjhjH5qzef7zUf4f62GbHZ
+g0Gbn8IVgYROTk/O72S23jAIsxLF9ZA/DQz3SBRbGynb8OEZQeBBps68YaStlVwk
+g04UeycHNuZGV/hB9IgGAwJkH/lbY24CQ5ikTzOfLhHCDdGXEkybJgJNOnOg+ceM
+pOfxzeb2zk3M4o66MA0aLMbM2amWvdUBopJKRfEwBC7erNVfZOsx1VnO+zSxgl9L
+Ae/CfvTpWoT11H5N8Qop7PoMa70V
+-----END CERTIFICATE-----
+-----BEGIN CERTIFICATE-----
+MIIC7TCCAdWgAwIBAgIJAIxu3kp0I0VqMA0GCSqGSIb3DQEBCwUAMBsxGTAXBgNV
+BAMTEFN5c3RlbURTIFRlc3QgQ0EwIBcNMjYwNzI5MTIyOTA4WhgPMjEyNjA3MDUx
+MjI5MDhaMBsxGTAXBgNVBAMTEFN5c3RlbURTIFRlc3QgQ0EwggEiMA0GCSqGSIb3
+DQEBAQUAA4IBDwAwggEKAoIBAQCr0ehP5tRAlvTEPNj0nx5D0sJTQRkdMCs3VERt
+dDRebaecIOZse5Uve4r5qKr/YyTnscFZvl4CeM04AKXljhcUPeea/e2eeQegh7IF
+bCPWJXcbTFe4LzJ89asZZEqV7BN0IfDU9OoGKyPILF/1UJ79e8N89KoIiHZYFl9o
+Lj4KlD3dWKRluWG6TWYxBYcnBt2D/cFXXaEa62rqoVp1hIQ/MF232V9eOBvYiCPs
+IYnhPAIjkYz8LhYp02azeULGD3mrck08AFeGjLuXHBVWsFkcM0nb2stOenhkKj8k
+MqbNVA8/iFPrrbkCLROffZnIa0JficXAR2dbe7eUXhHnCpt1AgMBAAGjMjAwMB0G
+A1UdDgQWBBQFgrHcX+3rTCqFWbekJ4qIrH68izAPBgNVHRMBAf8EBTADAQH/MA0G
+CSqGSIb3DQEBCwUAA4IBAQCfHThIA+XEdpP3a/ySBNS2yrsrKe1G6JDRdzMNQiQv
+t1ymTQ1Dm7bDUo7L1fFGX/cMjCo3+2Q5owIGO4t9DYJ6cpSBAad1HLNVmQOTtIeD
+/IXwCpODWe4ZL/zcuJcCbidLuEsy56fLNeuf2fkkWCaGL6ehpHsp09MFwgrU7xLa
+MpPBchHsKmbx8QbpQBYpMSGKvS9uTtrRyEy6OKyyXJfsfR6GlJbVH1cgUN+eg0cp
+PiCbrd6PyeobawteRrfJkCqKvh6/nvdSHXnkfmz47RMC2Obs41LX43Boh5UdHw65
+aJfjh7O8CCHALCWHkvlJOfbEI6s1fj79H98r1//ifY+8
+-----END CERTIFICATE-----
diff --git a/src/test/resources/cert/localhost-key.pem b/src/test/resources/cert/localhost-key.pem
new file mode 100644
index 0000000000..850a43a06a
--- /dev/null
+++ b/src/test/resources/cert/localhost-key.pem
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDmPrqJLuzMTamy
+Bsth4vFOVF1J5eEyuhRUCE7SNWjn0lp9zdLex2s6dx/UhtS3lOrwIJWnZ0qZEfRN
+idZUWwePxonM7PjN9NeOUVJssxY1/riQszsXQXwrGUieZecwk2sFgNB759XDu2PH
+t4GFqGl8u22LtxBkcXyZaKsiLc3h9FxNRBm0u6Owu0v/DwmXp5/8xnrOjb1GqDxK
+yBW1lJbPSb+lLw1++shjsj/aVuLFpX0eVAN8Jqk4ObCjnBFJAUeDg9zKRmEVSypX
+qc5wKarB9oyUUiF1VFPK9hg7mQN2Cq/I4lc1jCAcIRyQdeI7LDB9tkA/OF7mREM6
+HyZE3FmtAgMBAAECggEAHky1/pSiy/YSdV+ohy6248B9cFqkqqjLQQ3A1a/6qLtJ
+dlHORMwIg+6mTTEbMDeUPVqEZz3UFtXCiSuw/XPnSFfvzXyH946XiV6RUsW0kBF/
+12cGyTYwcXmH0XSGmqFjzZsYlJ27R2FTLbasAFtb2nLN5TuHmDhJFeUs1Dgj5m6i
+gl9gUjAMXqKoyibU/EQsfK7Jrb+++RIr12O0TmsOhsaklRWBHbS9Amq8XTKDCfnl
+R08fpwn+sKAetA9q/rrx2oN1Jd9Hmvgii9BiCtT3L3RLwqKWv5EnsDRKQI1Y88cb
+VZA3YTU5OskBFRCiJSbcPr48ejYNztLzpNLaBdyT0QKBgQDrMHLJU+6SFtyEsLqh
+oadZEtJYhlVMtLg1SH7ro2CoRyaK3E8RPIE3i80pPri8Ah7Wy1bD6/Fn6dYE6a+W
+VjyV+Uf7dIPpqgMIlsffcAri+PAY0dZmGu8/iLPauWQiteDK68b6PANGM5FZk/d+
+ccHR7mx0OKwV3HVVaeWhRuMd5QKBgQD6nkhs9fnq+mm2I9mTA1a8V/cP0XnxeTFR
+RhC3bqOwf+Fyrnv6UvQEIIWH4TaFfz4RR8WjeJdGqQ+z+Qak4sVjealgJTNZ5F2u
+YKrn4pIZDzIyyTI19+y8tp3EywQFEF1zKItfi2h0RiUOPsTo+vD0YYhcqC9oMbHE
+ODBpUkFQKQKBgQCWSCYA2Z3nQa51J0yKPXZmp207XdMhqZTPj1xyi7oWrShGsNHh
+LK1Q5gcZpNd8Y0p7bAEsPhbKlJPKHdyyDra2Ckzhs6ka5ST9FwPulXSPZgxdf7Al
+HG7mRR7P04jV2Swj3hcODMz2zbrB55fM9zmnQFeSyCfF7FIZWwp9TIORtQKBgFed
+/rQZSsZbxZln7yj2gdxW5IkjMv644AUJ+c4nYBLUonz1g2KAnc7Tj9txYR5K3egs
+r2v3POv3LwY8iZYbseaVIiH633kN3bKZGSb4jxsztNkMfgFgK+PN9FpYn48lqYYZ
+JqDAnEQKQeo5B55sHNFTR9kc83X56awv+LzZhPwBAoGALOuOGgxpXNWogMromKTp
+zNJt3MmSaJttWIq+peCWq5rinNIwvolYEQIJF3KDmvq7NPHBStEa5LZUVLUEyRO+
+hpBa4vmWmM7BSieVrer7P60zFr3fKoPh8+6wEHkmSCBpHel2iffU2zIcKuqQzSqi
+1h2tg7p4ZKqHjislPnaznMc=
+-----END PRIVATE KEY-----
diff --git a/src/test/resources/cert/otherhost-cert.pem b/src/test/resources/cert/otherhost-cert.pem
new file mode 100644
index 0000000000..71f9287eed
--- /dev/null
+++ b/src/test/resources/cert/otherhost-cert.pem
@@ -0,0 +1,37 @@
+-----BEGIN CERTIFICATE-----
+MIIDGzCCAgOgAwIBAgIIOj85LBlEMC8wDQYJKoZIhvcNAQELBQAwGzEZMBcGA1UE
+AxMQU3lzdGVtRFMgVGVzdCBDQTAgFw0yNjA3MjkxMjI5MTFaGA8yMTI2MDcwNTEy
+MjkxMVowHDEaMBgGA1UEAxMRb3RoZXIuZXhhbXBsZS5jb20wggEiMA0GCSqGSIb3
+DQEBAQUAA4IBDwAwggEKAoIBAQC9/pR8Tu2JQTGasc7yJrRqIjLF5KLj0ysUeEgi
+lS0i3sC/hbe165ckbxiKH38+ipiU/h9L+Z5nuLFuQc8RW3ER9TPcvYnIdRtLpq1j
+eUGhqAxj3ZgR1MPrpXNJGZRUgAWuGwCCRDFuHdFSWfVh3XF2hpxVIpWgP6dZ4vku
+eB2M23d4xyIMCuZ124q56QeOc6mPR0OTGsjn5/H1JFOLNGoZ5xaUcMcIKVs2gosm
+1opKI5x3Ty/v1Pdqemj3zBPp1asvzVCn0zF/pnV0MNKQWlrQMiyhWg51f6QNasei
+KwpguLp8d1Lt1Q0yLyBoAsf8trK39j95vukyYT2abxLEwcK9AgMBAAGjYDBeMB0G
+A1UdDgQWBBT8kGblKRiZZ5Jitaq5QFfXbAAdUTAcBgNVHREEFTATghFvdGhlci5l
+eGFtcGxlLmNvbTAfBgNVHSMEGDAWgBQFgrHcX+3rTCqFWbekJ4qIrH68izANBgkq
+hkiG9w0BAQsFAAOCAQEAFBOoWyq7NxMAatcgakb1ldOr7HImXBdfgno2B4Hi/R3e
+NBUkXCdEoO0FBFDu2qcOgXqIv1d62RCTzgcWcmQM8Iyx6zVmgyIRB2QFIrgHd3F+
+ndO59vz5CsAW8lZ6yDtLTMvM8tjPZrHXgkqpUXGry0wTFBodrYuHTFpUeArmeJ74
+r1kGyrc0P5+ZVeo+PI6rGZRr6prne7sdNHROSLcPWtYyaSIROlEML7mAq1TbzCH1
+Dh+6F89G8mEQG/JPSB3/b3kPFQnm+4VgT1KajWOnT67gqUFI69iBh/DzdqERWkfB
+ljUrfnZokHKEaoAZFH3I0B+3dmI2tg1YVzPKjfkA2g==
+-----END CERTIFICATE-----
+-----BEGIN CERTIFICATE-----
+MIIC7TCCAdWgAwIBAgIJAIxu3kp0I0VqMA0GCSqGSIb3DQEBCwUAMBsxGTAXBgNV
+BAMTEFN5c3RlbURTIFRlc3QgQ0EwIBcNMjYwNzI5MTIyOTA4WhgPMjEyNjA3MDUx
+MjI5MDhaMBsxGTAXBgNVBAMTEFN5c3RlbURTIFRlc3QgQ0EwggEiMA0GCSqGSIb3
+DQEBAQUAA4IBDwAwggEKAoIBAQCr0ehP5tRAlvTEPNj0nx5D0sJTQRkdMCs3VERt
+dDRebaecIOZse5Uve4r5qKr/YyTnscFZvl4CeM04AKXljhcUPeea/e2eeQegh7IF
+bCPWJXcbTFe4LzJ89asZZEqV7BN0IfDU9OoGKyPILF/1UJ79e8N89KoIiHZYFl9o
+Lj4KlD3dWKRluWG6TWYxBYcnBt2D/cFXXaEa62rqoVp1hIQ/MF232V9eOBvYiCPs
+IYnhPAIjkYz8LhYp02azeULGD3mrck08AFeGjLuXHBVWsFkcM0nb2stOenhkKj8k
+MqbNVA8/iFPrrbkCLROffZnIa0JficXAR2dbe7eUXhHnCpt1AgMBAAGjMjAwMB0G
+A1UdDgQWBBQFgrHcX+3rTCqFWbekJ4qIrH68izAPBgNVHRMBAf8EBTADAQH/MA0G
+CSqGSIb3DQEBCwUAA4IBAQCfHThIA+XEdpP3a/ySBNS2yrsrKe1G6JDRdzMNQiQv
+t1ymTQ1Dm7bDUo7L1fFGX/cMjCo3+2Q5owIGO4t9DYJ6cpSBAad1HLNVmQOTtIeD
+/IXwCpODWe4ZL/zcuJcCbidLuEsy56fLNeuf2fkkWCaGL6ehpHsp09MFwgrU7xLa
+MpPBchHsKmbx8QbpQBYpMSGKvS9uTtrRyEy6OKyyXJfsfR6GlJbVH1cgUN+eg0cp
+PiCbrd6PyeobawteRrfJkCqKvh6/nvdSHXnkfmz47RMC2Obs41LX43Boh5UdHw65
+aJfjh7O8CCHALCWHkvlJOfbEI6s1fj79H98r1//ifY+8
+-----END CERTIFICATE-----
diff --git a/src/test/resources/cert/otherhost-key.pem b/src/test/resources/cert/otherhost-key.pem
new file mode 100644
index 0000000000..d85091b938
--- /dev/null
+++ b/src/test/resources/cert/otherhost-key.pem
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQC9/pR8Tu2JQTGa
+sc7yJrRqIjLF5KLj0ysUeEgilS0i3sC/hbe165ckbxiKH38+ipiU/h9L+Z5nuLFu
+Qc8RW3ER9TPcvYnIdRtLpq1jeUGhqAxj3ZgR1MPrpXNJGZRUgAWuGwCCRDFuHdFS
+WfVh3XF2hpxVIpWgP6dZ4vkueB2M23d4xyIMCuZ124q56QeOc6mPR0OTGsjn5/H1
+JFOLNGoZ5xaUcMcIKVs2gosm1opKI5x3Ty/v1Pdqemj3zBPp1asvzVCn0zF/pnV0
+MNKQWlrQMiyhWg51f6QNaseiKwpguLp8d1Lt1Q0yLyBoAsf8trK39j95vukyYT2a
+bxLEwcK9AgMBAAECggEAF+o31AbWUPDHFub7OtFC49oepHixRTaTJV43jDzVQ96g
+iesBsxEuwuwF/XrV+DAXYSe0lkpbFUiy8sMvVoq5So6gAtDLy1LsRuM5z3vXlkrS
+Fm7x4Yqzx5FZl9GzsUg1DtOAxqThSPBRZQmEQNeQHOB4RKIYDeX9QWv3vBDr/Ur5
+qAwekZR8voIT/RJMdbLngNlmwEL6xWromOFz2GxF4Cb86F9/HFKg3rP7FjUIpR/Z
+rlha4zN2mMexgDIHpvrng/sWkwC+sOexIK3lkHgp2nvIeD3GVXnmY5dOh0NDdoFL
+3L5l+8lKaADq96kELL9Txf1snBrGtd8H0LO6UfEyEQKBgQC/wZMQm21COvz1zBaw
+0hMCRQC1VFt2AYrQE8pmnv70hIdpAHvHWh+sDfrDN1zwZwcbHlFF4I1kZTF+mEsd
+z4UDQUk/2wD/2QZ77jB5QaeuLBe16S6ULbIi67LVhWFAR5i5ILYjNmhRJtz0Ve5K
+E4/oNLsNRkCIMmbJmZ7mRGjOjQKBgQD9pejNKylzghKd3KTGAVRhoOBiPC4SgGYd
+sIjdI3GxbuWbfW4RNh5QtNvrFxPJOSiZRKIMjR+gG+wYMGzxIx5zyLrWXqDIf4zR
+iJOMAw4uWN/1wrCzVbi7X5eu4IVG3cIkWNZSFnSGkKUOBTeiW9WZmjKG6bVMOfbJ
+EsG6K4iQ8QKBgEV9mxQbn16vDdjtmxN9LdJWu0j7RyHesTVy1piV6gMmvAO7XyAB
+cxThBA0W1SFx1MtpEz7lf5fwbB1ah25INAXX9PmlHhmZxpXG3d4zgtbFt9n+pRih
+7rpk/CwQ6AtpZtlAF4FvSCKQmOYa9f32VOJrqZXH7b7ttP4+I62DARBJAoGAMZwL
+hmVUvCTKo1mOWLPV3ypp+Iywrimyz0fB3Q6bpAp+mgTUTEV7dGmLQdXHpumpCSEl
+WLMZZmVPrgN6q0clI5w0/syPQefAkRLXWOEYGvSDCTxE9y5i7TLrJeb/6jZhTF6b
+vH5r2A3eWnmmwfiYNGy2STDYpsoHfJhQj6sIEOECgYAZx4mGB1MC7WcDtrZW2Z+M
+OZlIAOgTtkVHJG1L0eESHAAngGWJeSE6ygOFlDUwYnL4jGH0TAEqGdd5NR/Iz0vE
+M9kZn55PeNrxukri3tCTKyTVp0vsIUfhtUzL46rA2xDPn3rKnnaE8fAqGQl1Vqxz
+ycVzx8wP/DN0GMKmualGJA==
+-----END PRIVATE KEY-----
diff --git a/src/test/resources/cert/untrusted-ca-cert.pem b/src/test/resources/cert/untrusted-ca-cert.pem
new file mode 100644
index 0000000000..902ad24f8a
--- /dev/null
+++ b/src/test/resources/cert/untrusted-ca-cert.pem
@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIIDADCCAeigAwIBAgIIJS61Qw8p3ygwDQYJKoZIhvcNAQELBQAwJTEjMCEGA1UE
+AxMaU3lzdGVtRFMgVW50cnVzdGVkIFRlc3QgQ0EwIBcNMjYwNzI5MTIyOTA5WhgP
+MjEyNjA3MDUxMjI5MDlaMCUxIzAhBgNVBAMTGlN5c3RlbURTIFVudHJ1c3RlZCBU
+ZXN0IENBMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAivWdwNG5FIPW
+T5ADOF0DtQwvPd9MbMBBL+xSD60hKbSYArofirH8T3Z4MtDc58e8rWIZuTvy4qcZ
+e2n+vAHZPyPWmHEjNsUGO1Vowd+QEIkzW0y+QTA2Z4KIAAAu1BO4o7mCnyMT34C+
+l3J9mK9TaBHm4vneYRHHuVMf70Y4cgBh9afgA9eCt3782SeSMTykLtvDvpBm3p2f
+YcPB7BNkVqv31Rel4Ru4veY71hZ6hzbmyS0eRQiTY39T072Oa8GAdud8ahTyfbre
+DcbK8MfKAIopCSVymSHa2LitQFdB1AAWCkZYDO8kXAvtRT5jeM6r4k4wTQ6cctGL
+g/XVXJb8zwIDAQABozIwMDAdBgNVHQ4EFgQUPpgtLYjF0t4Ze1cnCKt0K+xbeLcw
+DwYDVR0TAQH/BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAQ+PeSvSVNTALOm11
+4W72YAOgfukTVVwnZGp3DKo1MSX9hW5vr/hpz+xknHoMdssscWWMtTvdXG2mimvM
+Ryf+RHGNaWvTLb6GQFL9XWtNLEpQed/eUg8mKj/HUM7s4CqitWX9HN7RuL1sI20b
+L+oFvb/q8wMlbzjCPQsqyuiJ8r2nWO63QSx6QMFj51/z38NbgW++pcB4cVUFh8Jp
+HHvCDuwgCKKAGSkxq020QbU190VA1bUVWmd7B/gfVQvgVJiIxTwT9pbFKp+ImgE0
+Hmjt5kZiRB3op52rBGySfphnEKCYbGRtoS00ry2gZ60isG4OkyCk4spPVrXtdVzd
+/pvgPQ==
+-----END CERTIFICATE-----
diff --git a/scripts/tutorials/federated/conf/ssl.xml b/src/test/scripts/functions/federated/io/config/OtherHostSSLConfig.xml
similarity index 58%
copy from scripts/tutorials/federated/conf/ssl.xml
copy to src/test/scripts/functions/federated/io/config/OtherHostSSLConfig.xml
index f375ba33fe..8cf52db7d1 100644
--- a/scripts/tutorials/federated/conf/ssl.xml
+++ b/src/test/scripts/functions/federated/io/config/OtherHostSSLConfig.xml
@@ -16,6 +16,14 @@
* specific language governing permissions and limitations
* under the License.
-->
+
<root>
+ <!-- The worker presents a certificate signed by the trusted authority, but issued for
+ other.example.com instead of the contacted localhost, therefore it has to be rejected. -->
+ <sysds.local.spark.number.threads>2</sysds.local.spark.number.threads>
<sysds.federated.ssl>true</sysds.federated.ssl>
-</root>
\ No newline at end of file
+ <sysds.federated.ssl.cert>src/test/resources/cert/otherhost-cert.pem</sysds.federated.ssl.cert>
+ <sysds.federated.ssl.key>src/test/resources/cert/otherhost-key.pem</sysds.federated.ssl.key>
+ <sysds.federated.ssl.trust>src/test/resources/cert/ca-cert.pem</sysds.federated.ssl.trust>
+ <sysds.federated.timeout>128</sysds.federated.timeout>
+</root>
diff --git a/scripts/tutorials/federated/conf/ssl.xml b/src/test/scripts/functions/federated/io/config/SignedSSLConfig.xml
similarity index 58%
copy from scripts/tutorials/federated/conf/ssl.xml
copy to src/test/scripts/functions/federated/io/config/SignedSSLConfig.xml
index f375ba33fe..702f014450 100644
--- a/scripts/tutorials/federated/conf/ssl.xml
+++ b/src/test/scripts/functions/federated/io/config/SignedSSLConfig.xml
@@ -16,6 +16,14 @@
* specific language governing permissions and limitations
* under the License.
-->
+
<root>
+ <!-- The worker presents a certificate issued for localhost and signed by the test authority,
+ which the coordinator trusts. Used by both sides of the accepted connection. -->
+ <sysds.local.spark.number.threads>2</sysds.local.spark.number.threads>
<sysds.federated.ssl>true</sysds.federated.ssl>
-</root>
\ No newline at end of file
+ <sysds.federated.ssl.cert>src/test/resources/cert/localhost-cert.pem</sysds.federated.ssl.cert>
+ <sysds.federated.ssl.key>src/test/resources/cert/localhost-key.pem</sysds.federated.ssl.key>
+ <sysds.federated.ssl.trust>src/test/resources/cert/ca-cert.pem</sysds.federated.ssl.trust>
+ <sysds.federated.timeout>128</sysds.federated.timeout>
+</root>
diff --git a/scripts/tutorials/federated/conf/ssl.xml b/src/test/scripts/functions/federated/io/config/UntrustedSSLConfig.xml
similarity index 67%
copy from scripts/tutorials/federated/conf/ssl.xml
copy to src/test/scripts/functions/federated/io/config/UntrustedSSLConfig.xml
index f375ba33fe..6a2d1d4dae 100644
--- a/scripts/tutorials/federated/conf/ssl.xml
+++ b/src/test/scripts/functions/federated/io/config/UntrustedSSLConfig.xml
@@ -16,6 +16,12 @@
* specific language governing permissions and limitations
* under the License.
-->
+
<root>
+ <!-- Coordinator side only: trusts an authority unrelated to the one that signed the worker
+ certificate, therefore the worker has to be rejected. -->
+ <sysds.local.spark.number.threads>2</sysds.local.spark.number.threads>
<sysds.federated.ssl>true</sysds.federated.ssl>
-</root>
\ No newline at end of file
+ <sysds.federated.ssl.trust>src/test/resources/cert/untrusted-ca-cert.pem</sysds.federated.ssl.trust>
+ <sysds.federated.timeout>128</sysds.federated.timeout>
+</root>
diff --git a/src/test/scripts/functions/federated/io/generate-certificates.sh b/src/test/scripts/functions/federated/io/generate-certificates.sh
new file mode 100755
index 0000000000..9ce4264d0d
--- /dev/null
+++ b/src/test/scripts/functions/federated/io/generate-certificates.sh
@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+#
+# Regenerates the certificates in src/test/resources/cert, used by the federated SSL tests. The generated
+# files are checked in, so this script only has to be run if a certificate has to be replaced, for instance
+# because it expired or because the used algorithms are no longer accepted.
+#
+# The certificates are only used by tests, they authenticate nothing outside of a test run.
+#
+# Usage: src/test/scripts/functions/federated/io/generate-certificates.sh
+
+set -euo pipefail
+# from src/test/scripts/functions/federated/io up to src/test, then into the certificate directory
+cd "$(dirname "$0")/../../../../resources/cert"
+
+PW=changeit
+# ~100 years, the certificates are checked in and should not have to be replaced because of expiry
+DAYS=36500
+
+rm -f ./*.pem ./*.p12 ./*.csr
+
+# The authority the coordinator trusts, the basic constraint marks it as allowed to sign other certificates.
+keytool -genkeypair -alias ca -dname "CN=SystemDS Test CA" -ext bc:c -keyalg RSA -keysize 2048 \
+ -validity $DAYS -storetype PKCS12 -keystore ca.p12 -storepass $PW -keypass $PW
+keytool -exportcert -alias ca -keystore ca.p12 -storepass $PW -rfc -file ca-cert.pem
+
+# A second authority, unrelated to the one above and unknown to the coordinator.
+keytool -genkeypair -alias ca -dname "CN=SystemDS Untrusted Test CA" -ext bc:c -keyalg RSA -keysize 2048 \
+ -validity $DAYS -storetype PKCS12 -keystore untrusted-ca.p12 -storepass $PW -keypass $PW
+keytool -exportcert -alias ca -keystore untrusted-ca.p12 -storepass $PW -rfc -file untrusted-ca-cert.pem
+
+# Worker certificates signed by the first authority. The common name and the subject alternative names have
+# to match the host the coordinator connects to, otherwise the host name verification rejects the worker.
+gen_worker() {
+ local name=$1 dname=$2 san=$3
+ keytool -genkeypair -alias worker -dname "$dname" -keyalg RSA -keysize 2048 -validity $DAYS \
+ -storetype PKCS12 -keystore "$name.p12" -storepass $PW -keypass $PW
+ keytool -certreq -alias worker -keystore "$name.p12" -storepass $PW -file "$name.csr"
+ keytool -gencert -alias ca -keystore ca.p12 -storepass $PW -infile "$name.csr" \
+ -outfile "$name-cert.pem" -rfc -validity $DAYS -ext "san=$san"
+ # the chain presented by the worker, leaf certificate first
+ cat ca-cert.pem >> "$name-cert.pem"
+ # the private key, unencrypted PKCS#8 as read by the worker
+ openssl pkcs12 -in "$name.p12" -nocerts -nodes -passin "pass:$PW" \
+ | sed -n '/BEGIN PRIVATE KEY/,/END PRIVATE KEY/p' > "$name-key.pem"
+}
+
+gen_worker localhost "CN=localhost" "dns:localhost,ip:127.0.0.1"
+gen_worker otherhost "CN=other.example.com" "dns:other.example.com"
+
+rm -f ./*.p12 ./*.csr
+echo "Generated:"
+ls -1 ./*.pem