Support for TLS with server cert

Author: AndrewJSchofield

README.md

@@ -22,11 +22,11 @@ Change directory into the `kafka-connect-mq-source` directory:
 cd kafka-connect-mq-source
 ```
 
-Copy the JAR file `allclient-9.0.0.1.jar` that you unpacked from the ZIP file earlier into the `kafka-connect-mq-source` directory.
+Copy the JAR file `com.ibm.mq.allclient.jar` that you unpacked from the ZIP file earlier into the `kafka-connect-mq-source` directory.
 
 Run the following command to create a local Maven repository containing just this file so that it can be used to build the connector:
 ```shell
-mvn deploy:deploy-file -Durl=file://local-maven-repo -Dfile=allclient-9.0.0.1.jar -DgroupId=com.ibm.mq -DartifactId=allclient -Dpackaging=jar -Dversion=9.0.0.1
+mvn deploy:deploy-file -Durl=file://local-maven-repo -Dfile=com.ibm.mq.allclient.jar -DgroupId=com.ibm.mq -DartifactId=allclient -Dpackaging=jar -Dversion=9.0.0.1
 ```
 
 Build the connector using Maven:
@@ -109,6 +109,15 @@ There are three basic converters built into Apache Kafka, with the likely useful
 In addition, there is another converter for the Avro format that is part of the Confluent Platform. This has not been tested with the MQ source connector at this time.
 
 
+## Security
+The connector supports authentication with user name and password and also connections secured with TLS using a server-side certificate. It does not currently support TLS mutual authentication with client-side certificates.
+
+### Setting up TLS using a server-side certificate
+To enable use of TLS, set the configuration 'mq.ssl.cipher.suite' to the name of the cipher suite which matches the CipherSpec in the SSLCIPH attribute of the MQ server-connection channel. Use the table of supported cipher suites for MQ 9.0.x [here](https://www.ibm.com/support/knowledgecenter/en/SSFKSJ_9.0.0/com.ibm.mq.dev.doc/q113220_.htm) as a reference. Note that the names of the CipherSpecs as used in the MQ configuration are not necessarily the same as the cipher suite names that the connector uses. The connector uses the JMS interface so it follows the Java conventions.
+
+You will need to put the public part of the queue manager's certificate in a JSSE truststore and then provide the location of the truststore as property to the Kafka Connect worker that you're using to run the connector.
+
+
 ## Configuration
 The configuration options for the MQ Source Connector are as follows:
 
@@ -121,19 +130,24 @@ The configuration options for the MQ Source Connector are as follows:
 | mq.user.name            | The user name for authenticating with the queue manager     | string  |               | User name                   |
 | mq.password             | The password for authenticating with the queue manager      | string  |               | Password                    |
 | mq.message.body.jms     | Whether to interpret the message body as a JMS message type | boolean | false         |                             |
+| mq.ssl.cipher.suite     | The name of the cipher suite for TLS connection             | string  |               | Blank or valid cipher suite |
 | topic                   | The name of the target Kafka topic                          | string  |               | Topic name                  |
 
 
 ## Future enhancements
-The first version of the connector is intentionally basic. The idea is to enhance it with additional features to make it more capable. Some possible future enhancements are:
-* TLS connections
+The connector is intentionally basic. The idea is to enhance it over time with additional features to make it more capable. Some possible future enhancements are:
+* TLS mutual authentication
 * Message key support
 * Configurable schema for MQ messages
 * JMX metrics
 * JSON parsing so that the JSON type information is supplied to the converter
 * Testing with the Confluent Platform Avro converter and Schema Registry
 
 
+## Issues and contributions
+For issues relating specifically to this connect, please use the [GitHub issue tracker](https://github.com/ibm-messaging/kafka-connect-mq-source/issues). If you do submit a Pull Request related to this connector, please indicate in the Pull Request that you accept and agree to be bound by the terms of the [IBM Contributor License Agreement](CLA.md).
+
+
 ## License
 Copyright 2017 IBM Corporation
 

config/mq-source.properties

@@ -39,6 +39,10 @@ mq.queue=
 # Whether to interpret the message body as a JMS message type (default false) - optional
 # mq.message.body.jms=
 
+# The name of the cipher suite for TLS connection (default blank, meaning do not use TLS) - optional
+# See https://www.ibm.com/support/knowledgecenter/en/SSFKSJ_9.0.0/com.ibm.mq.dev.doc/q113220_.htm for valid values
+# mq.ssl.cipher.suite=
+
 # The name of the target Kafka topic - required
 topic=
 

src/main/java/com/ibm/mq/kafkaconnect/JMSReader.java

@@ -49,6 +49,7 @@ public class JMSReader {
     // Configs
     private String userName;
     private String password;
+    private String sslCipherSuite;
     private String topic;
     private boolean messageBodyJms;
 
@@ -123,6 +124,20 @@ public void setMessageBodyJms(boolean messageBodyJms)
         }
     }
 
+    /**
+     * Setter for SSL-related configuration.
+     * 
+     * @param sslCipherSuite     
+     */
+    public void setSSLConfiguration(String sslCipherSuite)
+    {
+        this.sslCipherSuite = sslCipherSuite;
+        if (this.sslCipherSuite != null)
+        {
+            mqConnFactory.setSSLCipherSuite(this.sslCipherSuite);
+        }
+    }
+
     /**
      * Connects to MQ.
      *
@@ -290,7 +305,8 @@ private void handleException(Throwable exc) throws ConnectException, RetriableEx
         while (t != null) {
             if (t instanceof MQException) {
                 MQException mqe = (MQException)t;
-                log.error("MQ error: CompCode {}, Reason {}", mqe.getCompCode(), mqe.getReason());
+                log.error("MQ error: CompCode {}, Reason {} {}", mqe.getCompCode(), mqe.getReason(),
+                          MQConstants.lookupReasonCode(mqe.getReason()));
                 reason = mqe.getReason();
                 break;
             }

src/main/java/com/ibm/mq/kafkaconnect/MQSourceConnector.java

@@ -63,6 +63,10 @@ public class MQSourceConnector extends SourceConnector {
     public static final String CONFIG_DOCUMENTATION_MQ_MESSAGE_BODY_JMS = "Whether to interpret the message body as a JMS message type.";
     public static final String CONFIG_DISPLAY_MQ_MESSAGE_BODY_JMS = "Message body as JMS";
 
+    public static final String CONFIG_NAME_MQ_SSL_CIPHER_SUITE = "mq.ssl.cipher.suite"; 
+    public static final String CONFIG_DOCUMENTATION_MQ_SSL_CIPHER_SUITE = "The name of the cipher suite for TLS (SSL) connection.";
+    public static final String CONFIG_DISPLAY_MQ_SSL_CIPHER_SUITE = "SSL cipher suite";
+
     public static final String CONFIG_NAME_TOPIC = "topic"; 
     public static final String CONFIG_DOCUMENTATION_TOPIC = "The name of the target Kafka topic.";
     public static final String CONFIG_DISPLAY_TOPIC = "Target Kafka topic";
@@ -158,6 +162,10 @@ public class MQSourceConnector extends SourceConnector {
                       CONFIG_DOCUMENTATION_MQ_MESSAGE_BODY_JMS, CONFIG_GROUP_MQ, 7, Width.SHORT,
                       CONFIG_DISPLAY_MQ_MESSAGE_BODY_JMS);
 
+        config.define(CONFIG_NAME_MQ_SSL_CIPHER_SUITE, Type.STRING, null, Importance.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_SSL_CIPHER_SUITE, CONFIG_GROUP_MQ, 8, Width.MEDIUM,
+                      CONFIG_DISPLAY_MQ_SSL_CIPHER_SUITE);
+
         config.define(CONFIG_NAME_TOPIC, Type.STRING, null, Importance.HIGH,
                       CONFIG_DOCUMENTATION_TOPIC, null, 0, Width.MEDIUM,
                       CONFIG_DISPLAY_TOPIC);

src/main/java/com/ibm/mq/kafkaconnect/MQSourceTask.java

@@ -36,6 +36,7 @@ public class MQSourceTask extends SourceTask {
     private String queueName;
     private String userName;
     private String password;
+    private String sslCipherSuite;
     private String topic;
 
     private static int BATCH_SIZE = 50;
@@ -69,11 +70,16 @@ public MQSourceTask() {
         queueName = props.get(MQSourceConnector.CONFIG_NAME_MQ_QUEUE);
         userName = props.get(MQSourceConnector.CONFIG_NAME_MQ_USER_NAME);
         password = props.get(MQSourceConnector.CONFIG_NAME_MQ_PASSWORD);
+        sslCipherSuite = props.get(MQSourceConnector.CONFIG_NAME_MQ_SSL_CIPHER_SUITE);
         topic = props.get(MQSourceConnector.CONFIG_NAME_TOPIC);
 
         // Construct a reader to interface with MQ
         reader = new JMSReader(queueManager, connectionNameList, channelName, queueName, userName, password, topic);
 
+        if (sslCipherSuite != null) {
+            reader.setSSLConfiguration(sslCipherSuite);
+        }
+
         String mbj = props.get(MQSourceConnector.CONFIG_NAME_MQ_MESSAGE_BODY_JMS);
         if (mbj != null) {
             reader.setMessageBodyJms(Boolean.parseBoolean(mbj));