Support SSL peer name

Author: AndrewJSchofield

README.md

@@ -110,13 +110,19 @@ In addition, there is another converter for the Avro format that is part of the
 
 
 ## 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.
+The connector supports authentication with user name and password and also connections secured with TLS using a server-side certificate and 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 the JSSE truststore used by the Kafka Connect worker that you're using to run the connector. If you need to specify extra arguments to the worker's JVM, you can use the EXTRA_ARGS environment variable.
 
+### Setting up TLS for mutual authentication
+You will need to put the public part of the client's certificate in the queue manager's key repository. You will also need to configure the worker's JVM with the location and password for the keystore containing the client's certificate.
+
+### Troubleshooting
+For troubleshooting, or to better understand the handshake performed by the IBM MQ Java client application in combination with your specific JSSE provider, you can enable debugging by setting `javax.net.debug=ssl` in the JVM environment.
+
 
 ## Configuration
 The configuration options for the MQ Source Connector are as follows:
@@ -130,18 +136,19 @@ 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 |
+| mq.ssl.cipher.suite     | The name of the cipher suite for TLS (SSL) connection       | string  |               | Blank or valid cipher suite |
+| mq.ssl.peer.name        | The distinguished name pattern of the TLS (SSL) peer        | string  |               | Blank or DN pattern         |
 | topic                   | The name of the target Kafka topic                          | string  |               | Topic name                  |
 
 
 ## Future enhancements
 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
+* Separate TLS configuration for the connector so that keystore location and so on can be specified as configurations
 
 
 ## Issues and contributions

config/mq-source.properties

@@ -39,10 +39,13 @@ 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
+# The name of the cipher suite for TLS (SSL) 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 distinguished name pattern of the TLS (SSL) peer - optional
+# mq.ssl.peer.name=
+
 # The name of the target Kafka topic - required
 topic=
 

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

@@ -50,6 +50,7 @@ public class JMSReader {
     private String userName;
     private String password;
     private String sslCipherSuite;
+    private String sslPeerName;
     private String topic;
     private boolean messageBodyJms;
 
@@ -127,14 +128,24 @@ public void setMessageBodyJms(boolean messageBodyJms)
     /**
      * Setter for SSL-related configuration.
      * 
-     * @param sslCipherSuite     
+     * @param sslCipherSuite     The name of the cipher suite for TLS (SSL) connection
+     * @param sslPeerName        The distinguished name pattern of the TLS (SSL) peer
      */
-    public void setSSLConfiguration(String sslCipherSuite)
+    public void setSSLConfiguration(String sslCipherSuite, String sslPeerName)
     {
         this.sslCipherSuite = sslCipherSuite;
         if (this.sslCipherSuite != null)
         {
             mqConnFactory.setSSLCipherSuite(this.sslCipherSuite);
+            if (this.sslPeerName != null)
+            {
+                try {
+                    mqConnFactory.setSSLPeerName(sslPeerName);
+                }
+                catch (JMSException jmse) {
+                    ;
+                }
+            }
         }
     }
 

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

@@ -64,9 +64,13 @@ public class MQSourceConnector extends SourceConnector {
     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_DOCUMENTATION_MQ_SSL_CIPHER_SUITE = "The name of the cipher suite for the TLS (SSL) connection.";
     public static final String CONFIG_DISPLAY_MQ_SSL_CIPHER_SUITE = "SSL cipher suite";
 
+    public static final String CONFIG_NAME_MQ_SSL_PEER_NAME = "mq.ssl.peer.name"; 
+    public static final String CONFIG_DOCUMENTATION_MQ_SSL_PEER_NAME = "The distinguished name pattern of the TLS (SSL) peer.";
+    public static final String CONFIG_DISPLAY_MQ_SSL_PEER_NAME = "SSL peer name";
+
     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";
@@ -166,6 +170,10 @@ public class MQSourceConnector extends SourceConnector {
                       CONFIG_DOCUMENTATION_MQ_SSL_CIPHER_SUITE, CONFIG_GROUP_MQ, 8, Width.MEDIUM,
                       CONFIG_DISPLAY_MQ_SSL_CIPHER_SUITE);
 
+        config.define(CONFIG_NAME_MQ_SSL_PEER_NAME, Type.STRING, null, Importance.MEDIUM,
+                      CONFIG_DOCUMENTATION_MQ_SSL_PEER_NAME, CONFIG_GROUP_MQ, 9, Width.MEDIUM,
+                      CONFIG_DISPLAY_MQ_SSL_PEER_NAME);
+
         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

@@ -37,6 +37,7 @@ public class MQSourceTask extends SourceTask {
     private String userName;
     private String password;
     private String sslCipherSuite;
+    private String sslPeerName;
     private String topic;
 
     private static int BATCH_SIZE = 50;
@@ -71,13 +72,14 @@ public MQSourceTask() {
         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);
+        sslPeerName = props.get(MQSourceConnector.CONFIG_NAME_MQ_SSL_PEER_NAME);
         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);
+            reader.setSSLConfiguration(sslCipherSuite, sslPeerName);
         }
 
         String mbj = props.get(MQSourceConnector.CONFIG_NAME_MQ_MESSAGE_BODY_JMS);