[SYSTEMDS-3955] Add signed SSL certificates support

.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

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>
 

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>

scripts/tutorials/federated/conf/ssl.xml

@@ -18,4 +18,13 @@
 -->
 <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, fill in the paths below 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>

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 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,
 			DEFAULT_FEDERATED_INITIALIZATION_TIMEOUT, FEDERATED_TIMEOUT, FEDERATED_MONITOR_FREQUENCY, FEDERATED_COMPRESSION,
 			ASYNC_PREFETCH, ASYNC_SPARK_BROADCAST, ASYNC_SPARK_CHECKPOINT, IO_COMPRESSION_CODEC
 		}; 

src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedData.java

@@ -242,6 +242,8 @@ private static ChannelInitializer<SocketChannel> createChannel(InetSocketAddress
 		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 synchronized static void createWorkGroup() {
 	}
 
 	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();
 		}
 

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 @@ 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());
+
+		// 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.info("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);

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<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();
+		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));
 			}
-			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<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);
-		}
+		};
 	}
 }