Support controlling use of MQCSP authentication

Author: Andrew Schofield

README.md

@@ -196,7 +196,7 @@ If you write your own RecordBuilder, you can access the MQMD fields of the MQ me
 
 
 ## Security
-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.
+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. You can also choose whether to use connection security parameters (MQCSP) depending on the security settings you're using in MQ.
 
 ### Setting up MQ connectivity using TLS with 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.1 [here](https://www.ibm.com/support/knowledgecenter/en/SSFKSJ_9.1.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.
@@ -234,8 +234,10 @@ The configuration options for the Kafka Connect source connector for IBM MQ are
 | mq.ssl.truststore.password   | The password of the JKS truststore to use for SSL (TLS) connections    | string  |                |                                                         |
 | mq.batch.size                | The maximum number of messages in a batch (unit of work)               | integer | 250            | 1 or greater                                            |
 | mq.message.mqmd.read         | Whether to enable reading of all MQMD fields                           | boolean | false          |                                                         |
+| mq.user.authentication.mqcsp | Whether to use MQ connection security parameters (MQCSP)               | boolean | true           |                                                         |
 | topic                        | The name of the target Kafka topic                                     | string  |                | Topic name                                              |
 
+
 ### Using a CCDT file
 Some of the connection details for MQ can be provided in a [CCDT file](https://www.ibm.com/support/knowledgecenter/en/SSFKSJ_9.1.0/com.ibm.mq.con.doc/q016730_.htm) by setting `mq.ccdt.url` in the MQ source connector configuration file. If using a CCDT file the `mq.connection.name.list` and `mq.channel.name` configuration options are not required.
 

config/mq-source.properties

@@ -45,6 +45,9 @@ mq.queue=
 # depending on the ConfigProvider implementation.
 # mq.password=${file:/var/run/secret.properties:secret-key}
 
+# Whether to use MQ connection security parameters (MQCSP) to provide credentials - optional
+# mq.user.authentication.mqcsp=
+
 # The CCDT URL to use to establish a client connection to the queue manager - optional
 # mq.ccdt.url=
 

src/main/java/com/ibm/eventstreams/connect/mqsource/JMSReader.java

@@ -107,6 +107,7 @@ public void configure(Map<String, String> props) {
         String sslKeystorePassword = props.get(MQSourceConnector.CONFIG_NAME_MQ_SSL_KEYSTORE_PASSWORD);
         String sslTruststoreLocation = props.get(MQSourceConnector.CONFIG_NAME_MQ_SSL_TRUSTSTORE_LOCATION);
         String sslTruststorePassword = props.get(MQSourceConnector.CONFIG_NAME_MQ_SSL_TRUSTSTORE_PASSWORD);
+        String useMQCSP = props.get(MQSourceConnector.CONFIG_NAME_MQ_USER_AUTHENTICATION_MQCSP);
         String topic = props.get(MQSourceConnector.CONFIG_NAME_TOPIC);
 
         int transportType = WMQConstants.WMQ_CM_CLIENT;
@@ -128,6 +129,9 @@ else if (connectionMode.equals(MQSourceConnector.CONFIG_VALUE_MQ_CONNECTION_MODE
             mqConnFactory.setTransportType(transportType);
             mqConnFactory.setQueueManager(queueManager);
             mqConnFactory.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, true);
+            if (useMQCSP != null) {
+                mqConnFactory.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, Boolean.parseBoolean(useMQCSP));
+            }
 
             if (transportType == WMQConstants.WMQ_CM_CLIENT) {
                 if (ccdtUrl != null) {

src/main/java/com/ibm/eventstreams/connect/mqsource/MQSourceConnector.java

@@ -119,6 +119,10 @@ public class MQSourceConnector extends SourceConnector {
     public static final String CONFIG_DOCUMENTATION_MQ_MESSAGE_MQMD_READ = "Whether to enable reading of all MQMD fields.";
     public static final String CONFIG_DISPLAY_MQ_MESSAGE_MQMD_READ = "Enable reading of MQMD fields";
 
+    public static final String CONFIG_NAME_MQ_USER_AUTHENTICATION_MQCSP = "mq.user.authentication.mqcsp";
+    public static final String CONFIG_DOCUMENTATION_MQ_USER_AUTHENTICATION_MQCSP = "Whether to use MQ connection security parameters (MQCSP).";
+    public static final String CONFIG_DISPLAY_MQ_USER_AUTHENTICATION_MQCSP = "User authentication using MQCSP";
+
     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";
@@ -281,6 +285,10 @@ public class MQSourceConnector extends SourceConnector {
                       CONFIG_DOCUMENTATION_MQ_MESSAGE_MQMD_READ, CONFIG_GROUP_MQ, 19, Width.SHORT,
                       CONFIG_DISPLAY_MQ_MESSAGE_MQMD_READ);
 
+        config.define(CONFIG_NAME_MQ_USER_AUTHENTICATION_MQCSP, Type.BOOLEAN, Boolean.TRUE, Importance.LOW,
+                      CONFIG_DOCUMENTATION_MQ_USER_AUTHENTICATION_MQCSP, CONFIG_GROUP_MQ, 20, Width.SHORT,
+                      CONFIG_DISPLAY_MQ_USER_AUTHENTICATION_MQCSP);
+
         config.define(CONFIG_NAME_TOPIC, Type.STRING, ConfigDef.NO_DEFAULT_VALUE, Importance.HIGH,
                       CONFIG_DOCUMENTATION_TOPIC, null, 0, Width.MEDIUM,
                       CONFIG_DISPLAY_TOPIC);