IBM MQ connect via TLS and MTLS

ibmmq_client_connect.py

#
# Manual tests to check generating the keystore and putting and getting
# a message.
#
import sys
import os

from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import (
    PrivateFormat,
    load_pem_private_key,
    pkcs12,
    )

# On Windows we need to setup the DLL path before importing ibmmq.
if os.name == 'nt':
    base_dll_dir = os.path.join(os.path.dirname(sys.executable), 'ibm-mq')
    os.add_dll_directory(base_dll_dir)

import ibmmq  # noqa:pycodestyle


# I don't know how to set the application name in another way.
os.environ['MQAPPLNAME'] = 'chevah-test'
os.environ['MQTCPTIMEOUT'] = '5'

def connect(conn_info, queue_manager, channel, ca_chain_certs, client_cert_and_key, cipher_spec=None, user='', password=''):
    print('Connecting to `{}` on `{}` via channel `{}`'.format(
        queue_manager, conn_info, channel))
    if cipher_spec is None:
        cipher_spec = b'ANY'

    # TLS config
    store_password = 'ignored'
    client_cert_name = b'client.cert'
    key_store_file = 'build-py3/ibm-mq-test-keystore.p12'

    if ca_chain_certs:
        encryption = (
            PrivateFormat.PKCS12.encryption_builder().
            kdf_rounds(50000).
            key_cert_algorithm(pkcs12.PBES.PBESv1SHA1And3KeyTripleDESCBC).
            hmac_hash(hashes.SHA1()).build(store_password.encode('ascii'))
            )
        cert = x509.load_pem_x509_certificate(
            open(client_cert_and_key, 'rb').read()
            )
        # Key has no password.
        key = load_pem_private_key(
            open(client_cert_and_key, 'rb').read(),
            None)
        ca_certs = []
        index = 0
        for ca_cert in x509.load_pem_x509_certificates(
            open(ca_chain_certs, 'rb').read()
                ):
            index += 1
            ca_certs.append(pkcs12.PKCS12Certificate(
                ca_cert, "ca-cert-{}".format(index).encode('ascii')))

        p12 = pkcs12.serialize_key_and_certificates(
            client_cert_name, key, cert, ca_certs, encryption)
        stream = open(key_store_file, 'wb')
        stream.write(p12)
        stream.close()

    cd = ibmmq.CD()
    cd.ChannelName = channel
    cd.ConnectionName = conn_info
    cd.ChannelType = ibmmq.CMQXC.MQCHT_CLNTCONN
    cd.TransportType = ibmmq.CMQXC.MQXPT_TCP

    cd.HeartbeatInterval = 2

    # Enable compression
    cd.MsgCompList[1] = ibmmq.CMQXC.MQCOMPRESS_ZLIBHIGH

    # https://www.ibm.com/docs/en/ibm-mq/9.4.x?topic=jms-tls-cipherspecs-ciphersuites-in-mq-classes
    if cipher_spec:
        cd.SSLCipherSpec = cipher_spec

    sco = ibmmq.SCO()
    if ca_chain_certs:
        sco.KeyRepository = key_store_file
        sco.KeyRepoPassword = store_password
        sco.CertificateLabel = client_cert_name
        # MQ_CERT_VAL_POLICY_NONE = 2
        sco.CertificateValPolicy = 2

    connect_options = ibmmq.CMQC.MQCNO_HANDLE_SHARE_BLOCK
    qmgr = ibmmq.QueueManager(None)
    qmgr.connect_with_options(
        queue_manager,
        user=user,
        password=password,
        cd=cd,
        sco=sco,
        opts=connect_options,
        )
    os.unlink(key_store_file) if os.path.exists(key_store_file) else None
    print('Connected to queue manager `{}` on channel `{}`'.format(
        queue_manager, channel))
    return qmgr