Fix up SNI function, while (re)allowing proxy support #573

Author: Ian Craggs

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/SSLNetworkModule.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2018 IBM Corp.
+ * Copyright (c) 2009, 2019 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -16,8 +16,12 @@
 package org.eclipse.paho.client.mqttv3.internal;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 
 import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.SNIHostName;
+import javax.net.ssl.SNIServerName;
 import javax.net.ssl.SSLParameters;
 import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.net.ssl.SSLSession;
@@ -129,6 +133,13 @@ public void start() throws IOException, MqttException {
 		// RTC 765: Set a timeout to avoid the SSL handshake being blocked indefinitely
 		socket.setSoTimeout(this.handshakeTimeoutSecs * 1000);
 
+		// SNI support.  Should be automatic under some circumstances - not all, apparently
+		SSLParameters sslParameters = new SSLParameters();
+		List<SNIServerName> sniHostNames = new ArrayList<SNIServerName>(1);
+		sniHostNames.add(new SNIHostName(host));
+		sslParameters.setServerNames(sniHostNames);
+		((SSLSocket)socket).setSSLParameters(sslParameters);
+
 		// If default Hostname verification is enabled, use the same method that is used with HTTPS
 		if(this.httpsHostnameVerificationEnabled) {
 			SSLParameters sslParams = new SSLParameters();

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/TCPNetworkModule.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2018 IBM Corp.
+ * Copyright (c) 2009, 2019 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -24,7 +24,6 @@
 import java.net.SocketAddress;
 
 import javax.net.SocketFactory;
-import javax.net.ssl.SSLSocketFactory;
 
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.logging.Logger;
@@ -71,15 +70,8 @@ public void start() throws IOException, MqttException {
 			// @TRACE 252=connect to host {0} port {1} timeout {2}
 			log.fine(CLASS_NAME,methodName, "252", new Object[] {host, Integer.valueOf(port), Long.valueOf(conTimeout*1000)});
 			SocketAddress sockaddr = new InetSocketAddress(host, port);
-			if (factory instanceof SSLSocketFactory) {
-				// SNI support
-				Socket tempsocket = new Socket();
-				tempsocket.connect(sockaddr, conTimeout*1000);
-				socket = ((SSLSocketFactory)factory).createSocket(tempsocket, host, port, true);
-			} else {
-				socket = factory.createSocket();
-				socket.connect(sockaddr, conTimeout*1000);
-			}
+			socket = factory.createSocket();
+			socket.connect(sockaddr, conTimeout*1000);
 			socket.setSoTimeout(1000);
 		}
 		catch (ConnectException ex) {

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/IMqttClient.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2018 IBM Corp.
+ * Copyright (c) 2009, 2019 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -335,21 +335,6 @@ public interface IMqttClient { //extends IMqttAsyncClient {
 	 * @param messageListener one callbacks to handle incoming messages
 	 * @throws MqttException if there was an error registering the subscription.
 	 */
-public void subscribe(String[] topicFilters, IMqttMessageListener messageListener) throws MqttException;
-
-	/**
-	 * Subscribe to a topic, which may include wildcards.
-	 *
-	 * @see #subscribe(String[], int[])
-	 *
-	 * @param topicFilter the topic to subscribe to, which can include wildcards.
-	 * @param qos the maximum quality of service at which to subscribe. Messages
-	 * published at a lower quality of service will be received at the published
-	 * QoS.  Messages published at a higher quality of service will be received using
-	 * the QoS specified on the subscribe.
-	 * @param messageListener a callback to handle incoming messages
-	 * @throws MqttException if there was an error registering the subscription.
-	 */
 public void subscribe(String topicFilter, int qos, IMqttMessageListener messageListener) throws MqttException;
 
 	/**
@@ -783,7 +768,7 @@ public interface IMqttClient { //extends IMqttAsyncClient {
 	 * Get a topic object which can be used to publish messages.
 	 * <p>An alternative method that should be used in preference to this one when publishing a message is:</p>
 	 * <ul>
-	 * <li>{@link MqttLegacyBlockingClient#publish(String, MqttMessage)} to publish a message in a blocking manner
+	 * <li>{@link MqttClient#publish(String, MqttMessage)} to publish a message in a blocking manner
 	 * <li>or use publish methods on the non-blocking client like {@link IMqttAsyncClient#publish(String, MqttMessage, Object, MqttActionListener)}
 	 * </ul>
 	 * <p>When building an application,

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2018 IBM Corp.
+ * Copyright (c) 2009, 2019 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -959,7 +959,7 @@ public String getCurrentServerURI() {
 	 * <ul>
 	 * <li>{@link MqttAsyncClient#publish(String, MqttMessage)} to publish a message
 	 * in a non-blocking manner or</li>
-	 * <li>{@link MqttLegacyBlockingClient#publish(String, MqttMessage)} to publish
+	 * <li>{@link MqttClient#publish(String, MqttMessage)} to publish
 	 * a message in a blocking manner</li>
 	 * </ul>
 	 * <p>

…tv5/client/MqttLegacyBlockingClient.java …lipse/paho/mqttv5/client/MqttClient.javaorg.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttLegacyBlockingClient.java renamed to org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttClient.java org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttLegacyBlockingClient.java renamed to org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttClient.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2015 IBM Corp.
+ * Copyright (c) 2009, 2019 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -76,7 +76,7 @@
  *
  * @see IMqttClient
  */
-public class MqttLegacyBlockingClient implements IMqttClient {
+public class MqttClient implements IMqttClient {
 
 	protected MqttAsyncClient aClient = null; // Delegate implementation to MqttAsyncClient
 	protected long timeToWait = -1; // How long each method should wait for action to complete
@@ -168,7 +168,7 @@ public class MqttLegacyBlockingClient implements IMqttClient {
 	 * @throws MqttException
 	 *             if any other problem was encountered
 	 */
-	public MqttLegacyBlockingClient(String serverURI, String clientId) throws MqttException {
+	public MqttClient(String serverURI, String clientId) throws MqttException {
 		this(serverURI, clientId, new MqttDefaultFilePersistence());
 	}
 
@@ -275,7 +275,7 @@ public MqttLegacyBlockingClient(String serverURI, String clientId) throws MqttEx
 	 * @throws MqttException
 	 *             if any other problem was encountered
 	 */
-	public MqttLegacyBlockingClient(String serverURI, String clientId, MqttClientPersistence persistence)
+	public MqttClient(String serverURI, String clientId, MqttClientPersistence persistence)
 			throws MqttException {
 		aClient = new MqttAsyncClient(serverURI, clientId, persistence);
 	}
@@ -386,7 +386,7 @@ public MqttLegacyBlockingClient(String serverURI, String clientId, MqttClientPer
 	 * @throws MqttException
 	 *             if any other problem was encountered
 	 */
-	public MqttLegacyBlockingClient(String serverURI, String clientId, MqttClientPersistence persistence,
+	public MqttClient(String serverURI, String clientId, MqttClientPersistence persistence,
 			ScheduledExecutorService executorService) throws MqttException {
 		aClient = new MqttAsyncClient(serverURI, clientId, persistence, null, executorService);
 	}
@@ -516,13 +516,26 @@ public void subscribe(MqttSubscription[] subscriptions) throws MqttException {
 		int[] grantedQos = tok.getGrantedQos();
 		for (int i = 0; i < grantedQos.length; ++i) {
 			subscriptions[i].setQos(grantedQos[i]);
-			;
 		}
 		if (grantedQos.length == 1 && subscriptions[0].getQos() == 0x80) {
 			throw new MqttException(MqttClientException.REASON_CODE_SUBSCRIBE_FAILED);
 		}
 	}
 
+	@Override
+	public void subscribe(String[] topicFilters, int[] qos) throws MqttException {
+		if (topicFilters.length != qos.length) {
+			throw new MqttException(MqttClientException.REASON_CODE_UNEXPECTED_ERROR);
+		}
+
+		MqttSubscription[] subscriptions = new MqttSubscription[topicFilters.length];
+		for (int i = 0; i < topicFilters.length; ++i) {
+			subscriptions[i] = new MqttSubscription(topicFilters[i], qos[i]);
+		}
+
+		this.subscribe(subscriptions);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * 
@@ -533,11 +546,6 @@ public void subscribe(String topicFilter, IMqttMessageListener messageListener)
 		this.subscribe(new String[] { topicFilter }, new int[] { 1 }, new IMqttMessageListener[] { messageListener });
 	}
 
-	@Override
-	public void subscribe(String[] topicFilters, IMqttMessageListener messageListener) throws MqttException {
-		this.subscribe(topicFilters, new IMqttMessageListener[] { messageListener });
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * 
@@ -562,10 +570,6 @@ public void subscribe(String topicFilter, int qos, IMqttMessageListener messageL
 		this.subscribe(new String[] { topicFilter }, new int[] { qos }, new IMqttMessageListener[] { messageListener });
 	}
 
-	@Override
-	public void subscribe(String[] topicFilters, int[] qos) throws MqttException {
-		this.subscribe(topicFilters, qos, new IMqttMessageListener[] {});
-	}
 
 	public void subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners)
 			throws MqttException {
@@ -673,7 +677,7 @@ public void setTimeToWait(long timeToWaitInMillis) throws IllegalArgumentExcepti
 	 * Return the maximum time to wait for an action to complete.
 	 * 
 	 * @return the time to wait
-	 * @see MqttLegacyBlockingClient#setTimeToWait(long)
+	 * @see MqttClient#setTimeToWait(long)
 	 */
 	public long getTimeToWait() {
 		return this.timeToWait;

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttClientPersistence.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2014 IBM Corp.
+ * Copyright (c) 2009, 2019 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -23,8 +23,8 @@
 /**
  * Represents a persistent data store, used to store outbound and inbound messages while they
  * are in flight, enabling delivery to the QoS specified. You can specify an implementation
- * of this interface using {@link MqttLegacyBlockingClient#MqttLegacyBlockingClient(String, String, MqttClientPersistence)},
- * which the {@link MqttLegacyBlockingClient} will use to persist QoS 1 and 2 messages.
+ * of this interface using {@link MqttClient#MqttLegacyBlockingClient(String, String, MqttClientPersistence)},
+ * which the {@link MqttClient} will use to persist QoS 1 and 2 messages.
  * <p>
  * If the methods defined throw the MqttPersistenceException then the state of the data persisted
  * should remain as prior to the method being called. For example, if {@link #put(String, MqttPersistable)}

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/alpha/MqttClientPersistence.java

@@ -15,7 +15,7 @@
  */
 package org.eclipse.paho.mqttv5.client.alpha;
 
-import org.eclipse.paho.mqttv5.client.MqttLegacyBlockingClient;
+import org.eclipse.paho.mqttv5.client.MqttClient;
 import org.eclipse.paho.mqttv5.common.MqttPersistable;
 import org.eclipse.paho.mqttv5.common.MqttPersistenceException;
 

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/SSLNetworkModule.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2018 IBM Corp.
+ * Copyright (c) 2009, 2019 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -16,8 +16,12 @@
 package org.eclipse.paho.mqttv5.client.internal;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 
 import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.SNIHostName;
+import javax.net.ssl.SNIServerName;
 import javax.net.ssl.SSLParameters;
 import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.net.ssl.SSLSession;
@@ -125,12 +129,18 @@ public void start() throws IOException, MqttException {
 		// RTC 765: Set a timeout to avoid the SSL handshake being blocked indefinitely
 		socket.setSoTimeout(this.handshakeTimeoutSecs * 1000);
 
-		// If default Hostname verification is enabled, use the same method that is used
-		// with HTTPS
+		// SNI support.  Should be automatic under some circumstances - not all, apparently
+		SSLParameters sslParameters = new SSLParameters();
+		List<SNIServerName> sniHostNames = new ArrayList<SNIServerName>(1);
+		sniHostNames.add(new SNIHostName(host));
+		sslParameters.setServerNames(sniHostNames);
+		((SSLSocket)socket).setSSLParameters(sslParameters);
+
+		// If default Hostname verification is enabled, use the same method that is used with HTTPS
 		if (this.httpsHostnameVerificationEnabled) {
 			SSLParameters sslParams = new SSLParameters();
 			sslParams.setEndpointIdentificationAlgorithm("HTTPS");
-			((SSLSocket) socket).setSSLParameters(sslParams);
+			((SSLSocket) socket).setSSLParameters(sslParams);	
 		}
 
 		((SSLSocket) socket).startHandshake();

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/TCPNetworkModule.java

@@ -24,7 +24,6 @@
 import java.net.SocketAddress;
 
 import javax.net.SocketFactory;
-import javax.net.ssl.SSLSocketFactory;
 
 import org.eclipse.paho.mqttv5.client.MqttClientException;
 import org.eclipse.paho.mqttv5.client.logging.Logger;
@@ -69,25 +68,12 @@ public TCPNetworkModule(SocketFactory factory, String host, int port, String res
 	public void start() throws IOException, MqttException {
 		final String methodName = "start";
 		try {
-//			InetAddress localAddr = InetAddress.getLocalHost();
-//			socket = factory.createSocket(host, port, localAddr, 0);
 			// @TRACE 252=connect to host {0} port {1} timeout {2}
 			log.fine(CLASS_NAME,methodName, "252", new Object[] {host, new Integer(port), new Long(conTimeout*1000)});
 			SocketAddress sockaddr = new InetSocketAddress(host, port);
-			if (factory instanceof SSLSocketFactory) {
-				// SNI support
-				Socket tempsocket = new Socket();
-				tempsocket.connect(sockaddr, conTimeout*1000);
-				socket = ((SSLSocketFactory)factory).createSocket(tempsocket, host, port, true);
-			} else {
-				socket = factory.createSocket();
-				socket.connect(sockaddr, conTimeout*1000);
-			}
+			socket = factory.createSocket();
+			socket.connect(sockaddr, conTimeout*1000);
 			socket.setSoTimeout(1000);
-
-			// SetTcpNoDelay was originally set ot true disabling Nagle's algorithm.
-			// This should not be required.
-//			socket.setTcpNoDelay(true);	// TCP_NODELAY on, which means we do not use Nagle's algorithm
 		}
 		catch (ConnectException ex) {
 			//@TRACE 250=Failed to create TCP socket