Merge pull request #94 from Findly-Inc/initial_ssl_support

SSL support for connecting to mongodb instances

src/main/java/org/elasticsearch/river/mongodb/MongoDBRiver.java

@@ -24,6 +24,10 @@
 
 import java.io.IOException;
 import java.net.UnknownHostException;
+import java.security.KeyManagementException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -34,6 +38,12 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 
+import javax.net.SocketFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.X509TrustManager;
+
 import org.bson.BasicBSONObject;
 import org.bson.types.BSONTimestamp;
 import org.bson.types.ObjectId;
@@ -77,6 +87,7 @@
 import com.mongodb.Mongo;
 import com.mongodb.MongoClient;
 import com.mongodb.MongoClientOptions;
+import com.mongodb.MongoClientOptions.Builder;
 import com.mongodb.MongoException;
 import com.mongodb.MongoInterruptedException;
 import com.mongodb.QueryOperators;
@@ -104,6 +115,8 @@ public class MongoDBRiver extends AbstractRiverComponent implements River {
 	public final static String PORT_FIELD = "port";
 	public final static String OPTIONS_FIELD = "options";
 	public final static String SECONDARY_READ_PREFERENCE_FIELD = "secondary_read_preference";
+	public final static String SSL_CONNECTION_FIELD = "ssl";
+	public final static String SSL_VERIFY_CERT_FIELD = "sslverifycertificate";
 	public final static String DROP_COLLECTION_FIELD = "drop_collection";
 	public final static String EXCLUDE_FIELDS_FIELD = "exclude_fields";
 	public final static String FILTER_FIELD = "filter";
@@ -164,6 +177,8 @@ public class MongoDBRiver extends AbstractRiverComponent implements River {
 	protected final String mongoLocalPassword;
 	protected final String mongoOplogNamespace;
 	protected final boolean mongoSecondaryReadPreference;
+	protected final boolean mongoUseSSL;
+	protected final boolean mongoSSLVerifyCertificate;
 
 	protected final String indexName;
 	protected final String typeName;
@@ -182,6 +197,7 @@ public class MongoDBRiver extends AbstractRiverComponent implements River {
 	// private final TransferQueue<Map<String, Object>> stream = new
 	// LinkedTransferQueue<Map<String, Object>>();
 	private final BlockingQueue<Map<String, Object>> stream;
+	private SocketFactory sslSocketFactory;
 
 	private Mongo mongo;
 	private DB adminDb;
@@ -250,7 +266,11 @@ public MongoDBRiver(final RiverName riverName,
 								.get(SECONDARY_READ_PREFERENCE_FIELD), false);
 				dropCollection = XContentMapValues.nodeBooleanValue(
 						mongoOptionsSettings.get(DROP_COLLECTION_FIELD), false);
-
+				mongoUseSSL = XContentMapValues.nodeBooleanValue(
+						mongoOptionsSettings.get(SSL_CONNECTION_FIELD), false);
+				mongoSSLVerifyCertificate = XContentMapValues.nodeBooleanValue(
+						mongoOptionsSettings.get(SSL_VERIFY_CERT_FIELD), true);
+
 				if (mongoOptionsSettings.containsKey(EXCLUDE_FIELDS_FIELD)) {
 					excludeFields = new HashSet<String>();
 					Object excludeFieldsSettings = mongoOptionsSettings
@@ -274,6 +294,8 @@ public MongoDBRiver(final RiverName riverName,
 				mongoSecondaryReadPreference = false;
 				dropCollection = false;
 				excludeFields = null;
+				mongoUseSSL = false;
+				mongoSSLVerifyCertificate = false;
 			}
 
 			// Credentials
@@ -377,6 +399,8 @@ public MongoDBRiver(final RiverName riverName,
 			script = null;
 			dropCollection = false;
 			excludeFields = null;
+			mongoUseSSL = false;
+			mongoSSLVerifyCertificate = false;
 		}
 		mongoOplogNamespace = mongoDb + "." + mongoCollection;
 
@@ -555,10 +579,16 @@ && getAdminDb().isAuthenticated()) {
 
 	private Mongo getMongoClient() {
 		if (mongo == null) {
-			// TODO: MongoClientOptions should be configurable
-			MongoClientOptions mco = MongoClientOptions.builder()
+
+			Builder builder = MongoClientOptions.builder()
 					.autoConnectRetry(true).connectTimeout(15000)
-					.socketTimeout(60000).build();
+					.socketKeepAlive(true).socketTimeout(60000);
+			if (mongoUseSSL){
+				builder.socketFactory(getSSLSocketFactory());
+			}
+
+			// TODO: MongoClientOptions should be configurable
+			MongoClientOptions mco = builder.build();
 			mongo = new MongoClient(mongoServers, mco);
 		}
 		return mongo;
@@ -595,6 +625,42 @@ public void close() {
 			indexerThread.interrupt();
 		}
 	}
+
+	private SocketFactory getSSLSocketFactory() {
+		if (sslSocketFactory != null)
+			return sslSocketFactory;
+
+		if (!mongoSSLVerifyCertificate) {
+			try {
+				final TrustManager[] trustAllCerts = new TrustManager[] { new X509TrustManager() {
+
+					@Override
+					public X509Certificate[] getAcceptedIssuers() {
+						return null;
+					}
+
+					@Override
+					public void checkServerTrusted(X509Certificate[] chain, String authType)
+							throws CertificateException {
+					}
+
+					@Override
+					public void checkClientTrusted(X509Certificate[] chain, String authType)
+							throws CertificateException {
+					}
+				}};
+				final SSLContext sslContext = SSLContext.getInstance( "SSL" );
+		        sslContext.init( null, trustAllCerts, new java.security.SecureRandom() );
+		        // Create an ssl socket factory with our all-trusting manager
+		        sslSocketFactory = sslContext.getSocketFactory();
+		        return sslSocketFactory;
+			} catch(Exception ex) {
+				logger.error("Unable to build ssl socket factory without certificate validation, using default instead.", ex);
+			}
+		}
+		sslSocketFactory = SSLSocketFactory.getDefault();
+		return sslSocketFactory;
+	}
 
 	private class Indexer implements Runnable {
 
@@ -963,7 +1029,16 @@ private boolean assignCollections() {
 
 		@Override
 		public void run() {
-			mongo = new MongoClient(mongoServers);
+			Builder builder = MongoClientOptions.builder()
+					.autoConnectRetry(true).connectTimeout(15000)
+					.socketKeepAlive(true).socketTimeout(60000);
+			if (mongoUseSSL){
+				builder.socketFactory(getSSLSocketFactory());
+			}
+
+			// TODO: MongoClientOptions should be configurable
+			MongoClientOptions mco = builder.build();
+			mongo = new MongoClient(mongoServers, mco);
 
 			if (mongoSecondaryReadPreference) {
 				mongo.setReadPreference(ReadPreference.secondaryPreferred());