Support for SSL connections (#301)

com.ibm.streamsx.messaging/com.ibm.streamsx.messaging.jms/JMSSink/JMSSink.xml

@@ -19,6 +19,16 @@ For the `JMSSink` operator, the following message classes are supported:
 map, stream, bytes, xml, wbe, wbe22, and empty.
 The type of message is specified as the value of the message_class attribute in the connection specifications document.
 
+# SSL Support
+
+The `JMSSink` operator provides support for SSL via these parameters: *sslConnection*, *keyStore*, *keyStorePassword* and *trustStore*. 
+When *sslConnection* is set to `true`, the *keyStore*, *keyStorePassword* and *trustStore* parameters must be set. 
+
+**Note:** The `JMSSink` operator configures SSL by setting the JVM system properties via calls to `System.property()`. 
+Java operators that are fused into the same PE share the same JVM. This implies that any other Java operators fused into the 
+same PE as the `JMSSink` operator will have these SSL properties set. If this is undesirable, then the `JMSSink` operator
+should be placed into it's own PE. 
+
 # Behavior in a consistent region
 
 The `JMSSink` operator can be an operator within the reachability graph of a operator-driven consistent region.
@@ -348,6 +358,46 @@ If the **maxMessageSendRetries** is specified, you must also specify a value for
         <type>rstring</type>
         <cardinality>1</cardinality>
       </parameter>
+      <parameter>
+        <name>sslConnection</name>
+        <description>This parameter specifies whether the operator should attempt to connect using SSL. If this parameter is specified, then the *keyStore*, *keyStorePassword* and *trustStore* parameters must also be specified. The default value is `false`.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>boolean</type>
+        <cardinality>1</cardinality>
+      </parameter>
+      <parameter>
+        <name>keyStore</name>
+        <description>This parameter specifies the path to the keyStore. If a relative path is specified, the path is relative to the application directory. The *sslConnection* parameter must be set to true for this parameter to have any effect.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>rstring</type>
+        <cardinality>1</cardinality>
+      </parameter>
+      <parameter>
+        <name>keyStorePassword</name>
+        <description>This parameter specifies the password for the keyStore given by the *keyStore* parameter. The *sslConnection* parameter must be set to `true` for this parameter to have any effect.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>rstring</type>
+        <cardinality>1</cardinality>
+      </parameter>
+      <parameter>
+        <name>trustStore</name>
+        <description>This parameter specifies the path to the trustStore. If a relative path is specified, the path is relative to the application directory. The *sslConnection* parameter must be set to true for this parameter to have any effect.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>rstring</type>
+        <cardinality>1</cardinality>
+      </parameter>
+      <parameter>
+        <name>trustStorePassword</name>
+        <description>This parameter specifies the password for the trustStore given by the *trustStore* parameter. The *sslConnection* parameter must be set to `true` for this parameter to have any effect.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>rstring</type>
+        <cardinality>1</cardinality>
+      </parameter>
     </parameters>
     <inputPorts>
       <inputPortSet>

com.ibm.streamsx.messaging/com.ibm.streamsx.messaging.jms/JMSSource/JMSSource.xml

@@ -19,6 +19,17 @@ or a topic and creates tuples from the read data.
 The `JMSSource` operator converts each WebSphere MQ or Apache Active MQ message to a separate tuple
 and sends it to the output stream. A single message is converted into a single tuple.
 
+# SSL Support
+
+The `JMSSource` operator provides support for SSL via these parameters: *sslConnection*, *keyStore*, *keyStorePassword* and *trustStore*. 
+When *sslConnection* is set to `true`, the *keyStore*, *keyStorePassword* and *trustStore* parameters must be set. 
+
+**Note:** The `JMSSource` operator configures SSL by setting the JVM system properties via calls to `System.property()`. 
+Java operators that are fused into the same PE share the same JVM. This implies that any other Java operators fused into the 
+same PE as the `JMSSource` operator will have these SSL properties set. If this is undesirable, then the `JMSSource` operator
+should be placed into it's own PE. 
+
+
 # Behavior in a consistent region
 
 The `JMSSource` operator can participate in a consistent region. The operator must be at the start of a consistent region.
@@ -343,6 +354,46 @@ This parameter must be greater than zero and must be set if the JMSSource operat
         <type>rstring</type>
         <cardinality>1</cardinality>
       </parameter>
+      <parameter>
+        <name>sslConnection</name>
+        <description>This parameter specifies whether the operator should attempt to connect using SSL. If this parameter is specified, then the *keyStore*, *keyStorePassword* and *trustStore* parameters must also be specified. The default value is `false`.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>boolean</type>
+        <cardinality>1</cardinality>
+      </parameter>
+      <parameter>
+        <name>keyStore</name>
+        <description>This parameter specifies the path to the keyStore. If a relative path is specified, the path is relative to the application directory. The *sslConnection* parameter must be set to true for this parameter to have any effect.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>rstring</type>
+        <cardinality>1</cardinality>
+      </parameter>
+      <parameter>
+        <name>keyStorePassword</name>
+        <description>This parameter specifies the password for the keyStore given by the *keyStore* parameter. The *sslConnection* parameter must be set to `true` for this parameter to have any effect.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>rstring</type>
+        <cardinality>1</cardinality>
+      </parameter>
+      <parameter>
+        <name>trustStore</name>
+        <description>This parameter specifies the path to the trustStore. If a relative path is specified, the path is relative to the application directory. The *sslConnection* parameter must be set to true for this parameter to have any effect.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>rstring</type>
+        <cardinality>1</cardinality>
+      </parameter>  
+      <parameter>
+        <name>trustStorePassword</name>
+        <description>This parameter specifies the password for the trustStore given by the *trustStore* parameter. The *sslConnection* parameter must be set to `true` for this parameter to have any effect.</description>
+        <optional>true</optional>
+        <expressionMode>AttributeFree</expressionMode>
+        <type>rstring</type>
+        <cardinality>1</cardinality>
+      </parameter>          
     </parameters>
     <inputPorts>
     </inputPorts>

com.ibm.streamsx.messaging/impl/java/src/com/ibm/streamsx/messaging/jms/JMSSink.java

@@ -8,6 +8,12 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
 import java.util.logging.Logger;
 
 import javax.jms.Message;
@@ -190,7 +196,61 @@ public void setnReconnectionAttempts(Metric nReconnectionAttempts) {
     // password property name stored in application configuration
     private String passwordPropName;
 
+    private String keyStore;
+
+    private String trustStore;
+
+    private String keyStorePassword;
+
+    private String trustStorePassword;
+
+    private boolean sslConnection;
 
+    public boolean isSslConnection() {
+		return sslConnection;
+	}
+
+    @Parameter(optional = true)
+    public void setSslConnection(boolean sslConnection) {
+		this.sslConnection = sslConnection;
+	}
+
+    @Parameter(optional = true)
+    public void setTrustStorePassword(String trustStorePassword) {
+		this.trustStorePassword = trustStorePassword;
+	}
+
+    public String getTrustStorePassword() {
+		return trustStorePassword;
+	}
+
+    public String getTrustStore() {
+		return trustStore;
+	}
+
+    @Parameter(optional = true)
+    public void setTrustStore(String trustStore) {
+		this.trustStore = trustStore;
+	}
+
+    public String getKeyStorePassword() {
+		return keyStorePassword;
+	}
+
+    @Parameter(optional = true)
+    public void setKeyStorePassword(String keyStorePassword) {
+		this.keyStorePassword = keyStorePassword;
+	}
+
+    public String getKeyStore() {
+		return keyStore;
+	}
+
+    @Parameter(optional = true)
+    public void setKeyStore(String keyStore) {
+		this.keyStore = keyStore;
+	}
+
 	public String getAppConfigName() {
 		return appConfigName;
 	}
@@ -504,7 +564,7 @@ public static void checkParameters(OperatorContextChecker checker) {
 		checker.checkDependentParameters("userPropName", "appConfigName", "passwordPropName");
 		checker.checkDependentParameters("passwordPropName", "appConfigName", "userPropName");
 	}
-
+	
 	@Override
 	public synchronized void initialize(OperatorContext context)
 			throws ParserConfigurationException, InterruptedException,
@@ -514,6 +574,18 @@ public synchronized void initialize(OperatorContext context)
 
 		JmsClasspathUtil.setupClassPaths(context);
 
+		// set SSL system properties
+		if(isSslConnection()) {
+			if(context.getParameterNames().contains("keyStore"))
+				System.setProperty("javax.net.ssl.keyStore", getAbsolutePath(getKeyStore()));				
+			if(context.getParameterNames().contains("keyStorePassword"))
+				System.setProperty("javax.net.ssl.keyStorePassword", getKeyStorePassword());				
+			if(context.getParameterNames().contains("trustStore"))
+				System.setProperty("javax.net.ssl.trustStore",  getAbsolutePath(getTrustStore()));			
+			if(context.getParameterNames().contains("trustStorePassword"))
+				System.setProperty("javax.net.ssl.trustStorePassword",  getTrustStorePassword());
+		}
+
 		consistentRegionContext = context.getOptionalContext(ConsistentRegionContext.class);
 
 		operatorUniqueID = context.getPE().getDomainId() + "_" + context.getPE().getInstanceId() + "_" + context.getPE().getJobId() + "_" + context.getName();
@@ -627,9 +699,22 @@ public synchronized void initialize(OperatorContext context)
 
 		// register for data governance
 		registerForDataGovernance(connectionDocumentParser.getProviderURL(), connectionDocumentParser.getDestination());
-
 	}
 
+	protected String getAbsolutePath(String filePath) {
+		if(filePath == null) 
+			return null;
+
+
+		Path p = Paths.get(filePath);
+		if(p.isAbsolute()) {
+			return filePath;
+		} else {
+			File f = new File (getOperatorContext().getPE().getApplicationDirectory(), filePath);
+			return f.getAbsolutePath();
+		}
+	}
+
 	private void registerForDataGovernance(String providerURL, String destination) {
 		logger.log(TraceLevel.INFO, "JMSSink - Registering for data governance with providerURL: " + providerURL
 				+ " destination: " + destination);

com.ibm.streamsx.messaging/impl/java/src/com/ibm/streamsx/messaging/jms/JMSSource.java

@@ -9,8 +9,13 @@
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import javax.jms.JMSException;
@@ -189,6 +194,61 @@ public void setnReconnectionAttempts(Metric nReconnectionAttempts) {
     // password property name stored in application configuration
     private String passwordPropName;
 
+    private String keyStore;
+
+    private String trustStore;
+
+    private String keyStorePassword;
+
+    private String trustStorePassword;
+
+    private boolean sslConnection;
+
+    public boolean isSslConnection() {
+		return sslConnection;
+	}
+
+    @Parameter(optional = true)
+    public void setSslConnection(boolean sslConnection) {
+		this.sslConnection = sslConnection;
+	}
+
+    public String getTrustStorePassword() {
+		return trustStorePassword;
+	}
+
+    @Parameter(optional = true)
+    public void setTrustStorePassword(String trustStorePassword) {
+		this.trustStorePassword = trustStorePassword;
+	}
+
+    public String getTrustStore() {
+		return trustStore;
+	}
+
+    @Parameter(optional = true)
+    public void setTrustStore(String trustStore) {
+		this.trustStore = trustStore;
+	}
+
+    public String getKeyStorePassword() {
+		return keyStorePassword;
+	}
+
+    @Parameter(optional = true)
+    public void setKeyStorePassword(String keyStorePassword) {
+		this.keyStorePassword = keyStorePassword;
+	}
+
+    public String getKeyStore() {
+		return keyStore;
+	}
+
+    @Parameter(optional = true)
+    public void setKeyStore(String keyStore) {
+		this.keyStore = keyStore;
+	}    
+
 
 	public String getAppConfigName() {
 		return appConfigName;
@@ -534,7 +594,7 @@ public static void checkParameters(OperatorContextChecker checker) {
 		checker.checkDependentParameters("userPropName", "appConfigName", "passwordPropName");
 		checker.checkDependentParameters("passwordPropName", "appConfigName", "userPropName");
 	}
-
+	
 	@Override
 	public synchronized void initialize(OperatorContext context)
 			throws ParserConfigurationException, InterruptedException,
@@ -547,6 +607,18 @@ public synchronized void initialize(OperatorContext context)
 
 		JmsClasspathUtil.setupClassPaths(context);
 
+		// set SSL system properties
+		if(isSslConnection()) {
+			if(context.getParameterNames().contains("keyStore"))
+				System.setProperty("javax.net.ssl.keyStore", getAbsolutePath(getKeyStore()));				
+			if(context.getParameterNames().contains("keyStorePassword"))
+				System.setProperty("javax.net.ssl.keyStorePassword", getKeyStorePassword());				
+			if(context.getParameterNames().contains("trustStore"))
+				System.setProperty("javax.net.ssl.trustStore",  getAbsolutePath(getTrustStore()));			
+			if(context.getParameterNames().contains("trustStorePassword"))
+				System.setProperty("javax.net.ssl.trustStorePassword",  getTrustStorePassword());
+		}
+
 		if(consistentRegionContext != null) {
 			crState = new JMSSourceCRState();
 		}
@@ -629,6 +701,19 @@ public synchronized void initialize(OperatorContext context)
 
 	}
 
+	protected String getAbsolutePath(String filePath) {
+		if(filePath == null) 
+			return null;
+
+		Path p = Paths.get(filePath);
+		if(p.isAbsolute()) {
+			return filePath;
+		} else {
+			File f = new File (getOperatorContext().getPE().getApplicationDirectory(), filePath);
+			return f.getAbsolutePath();
+		}
+	}
+
 	private void registerForDataGovernance(String providerURL, String destination) {
 		logger.log(TraceLevel.INFO, "JMSSource - Registering for data governance with providerURL: " + providerURL
 				+ " destination: " + destination);

com.ibm.streamsx.messaging/info.xml

@@ -680,7 +680,7 @@ The <attribute> element has three possible attributes:
      * composite types
      * xml
 </info:description>
-    <info:version>5.1.0</info:version>
+    <info:version>5.1.1</info:version>
     <info:requiredProductVersion>4.2.0.0</info:requiredProductVersion>
   </info:identity>
   <info:dependencies/>