diff --git a/.gitignore b/.gitignore
index acb9435ac64..590413bb8f9 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 153dcb6ef2d..8d9db3fea8a 100644
--- a/conf/SystemDS-config.xml.template
+++ b/conf/SystemDS-config.xml.template
@@ -146,6 +146,21 @@
none
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
15
diff --git a/pom.xml b/pom.xml
index be50b05a92c..2560a38b7b7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -734,6 +734,8 @@
**/*.mtx
**/*.mtd
**/*.out
+
+ src/test/resources/cert/*.pem
**/__pycache__/**
**/part-*
**/*.keep
diff --git a/scripts/tutorials/federated/conf/ssl.xml b/scripts/tutorials/federated/conf/ssl.xml
index f375ba33fed..fdd05c6654b 100644
--- a/scripts/tutorials/federated/conf/ssl.xml
+++ b/scripts/tutorials/federated/conf/ssl.xml
@@ -18,4 +18,11 @@
-->
true
+
+
\ 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 d114ccf69b9..c01ba563747 100644
--- a/src/main/java/org/apache/sysds/conf/DMLConfig.java
+++ b/src/main/java/org/apache/sysds/conf/DMLConfig.java
@@ -125,6 +125,11 @@ 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_KEY_PASSWORD = "sysds.federated.ssl.keyPassword"; // Password of the worker private key, only required if the key is encrypted
+ 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 FEDERATED_SSL_HOSTNAME_VERIFICATION = "sysds.federated.ssl.hostnameVerification"; // boolean: verify that the worker certificate matches the contacted host
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 +216,11 @@ 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_KEY_PASSWORD, null);
+ _defaultVals.put(FEDERATED_SSL_TRUST, null);
+ _defaultVals.put(FEDERATED_SSL_HOSTNAME_VERIFICATION, "true");
_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 +485,7 @@ public String getConfigInfo() {
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, FEDERATED_SSL_HOSTNAME_VERIFICATION,
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 19277ba0843..52ff2ff9572 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 @@ private static ChannelInitializer createChannel(InetSocketAddress
DataRequestHandler handler) {
final int timeout = ConfigurationManager.getFederatedTimeout();
final boolean ssl = ConfigurationManager.isFederatedSSL();
+ if(ssl)
+ FederatedSSLUtil.SslConstructor();
return new ChannelInitializer<>() {
@Override
@@ -309,20 +311,35 @@ public synchronized static void createWorkGroup() {
private static class DataRequestHandler extends ChannelInboundHandlerAdapter {
private Promise _prom;
+ // A failure observed before the promise was assigned, e.g., a rejected SSL handshake
+ private Throwable _failure;
public DataRequestHandler() {
}
- public void setPromise(Promise prom) {
+ public synchronized void setPromise(Promise prom) {
_prom = prom;
+ if(_failure != null)
+ _prom.tryFailure(_failure);
}
@Override
- public void channelRead(ChannelHandlerContext ctx, Object msg) {
+ public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) {
_prom.setSuccess((FederatedResponse) msg);
ctx.close();
}
+ @Override
+ public synchronized 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.
+ if(_prom != null)
+ _prom.tryFailure(cause);
+ else
+ _failure = cause;
+ ctx.close();
+ }
+
public Promise getProm() {
return _prom;
}
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 f1300ef0f4a..73e7ffd6eee 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,25 @@
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);
private FederatedSSLUtil(){
// private constructor.
@@ -40,24 +46,95 @@ private 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());
+
+ if(man.verifyHostname) {
+ 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.
+ *
+ * @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 = conf.getTextValue(DMLConfig.FEDERATED_SSL_KEY_PASSWORD);
+
+ 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;
+ // Verify that the certificate of the contacted worker was issued for that host
+ protected final boolean verifyHostname;
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.info("Federated SSL trusting certificates in: " + trustPath);
+ context = SslContextBuilder.forClient()
+ .trustManager(readableFile(trustPath, DMLConfig.FEDERATED_SSL_TRUST)).build();
+ verifyHostname = conf.getBooleanValue(DMLConfig.FEDERATED_SSL_HOSTNAME_VERIFICATION);
}
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 fc8989053bc..302e6d1cfe5 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.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")
@@ -192,47 +186,30 @@ protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out)
}
private ChannelInitializer 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();
+ 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> 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));
}
- else {
- cert = null;
- cont2 = null;
- }
-
- return new ChannelInitializer<>() {
- @Override
- public void initChannel(SocketChannel ch) {
- final ChannelPipeline cp = ch.pipeline();
- if(sslEnabled)
- cp.addLast(cont2.newHandler(ch.alloc()));
-
- final Optional> 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 5f5c09e07cb..c841f73f4d4 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,47 +18,60 @@
*/
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.Before;
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() {
@@ -66,84 +79,188 @@ public void setUp() {
addTestConfiguration(TEST_NAME, new TestConfiguration(TEST_CLASS_DIR, TEST_NAME));
}
- @Parameterized.Parameters
- public static Collection