mTLS Setup

Overview

Standard TLS only proves the server's identity — the server itself accepts any client that knows the address and client ID. Mutual TLS (mTLS) closes that gap: the client also presents a certificate, so the server verifies who is connecting before accepting the connection. That matters on zero-trust networks and in regulated environments where "reaches the port" isn't an acceptable authorization model — the certificate becomes the credential.

TLSConfig with cert_file, key_file, and ca_file wires in three artifacts at client construction: the CA certificate (to verify the server, same as one-way TLS) plus the client's own certificate and private key (for the server to verify in return). Verification happens during the handshake, before any messaging traffic flows.

Gotchas: the certificate and key must be a matched pair signed by a CA the server trusts — a mismatch fails the handshake outright; all three files must be valid, unexpired PEM, and expiry breaks connections with no warning; and the CA that signed the client cert isn't necessarily the CA that verifies the server — mixing them up causes "works with TLS, fails with mTLS" confusion.

Prerequisites

• KubeMQ server running with mTLS enabled
• Python SDK installed (pip install kubemq)
• TLS certificates (client certificate, client key, and CA certificate)

Code

"""Example: Mutual TLS (mTLS) — client and server authenticate each other."""

from __future__ import annotations

import asyncio

from kubemq import TLSConfig
from kubemq import AsyncPubSubClient, EventMessage


async def main() -> None:
    # Mutual TLS: both client cert+key and CA cert are provided
    tls_config = TLSConfig(
        enabled=True,
        cert_file="/path/to/client-cert.pem",
        key_file="/path/to/client-key.pem",
        ca_file="/path/to/ca.pem",
    )

    async with AsyncPubSubClient(
        address="kubemq-server:50000",
        client_id="python-tls-mtls-setup-client",
        tls=tls_config,
    ) as client:
        info = await client.ping()
        print(f"Connected via mTLS to {info.host}")

        await client.publish_event(
            EventMessage(
                channel="python-tls.mtls-setup",
                body=b"mTLS secured message",
            )
        )
        print("Message sent over mTLS connection")


if __name__ == "__main__":
    asyncio.run(main())

How It Works

• TLSConfig with cert_file, key_file, and ca_file enables mutual TLS authentication.
• The server verifies the client's certificate, and the client verifies the server's certificate, establishing bidirectional trust.
• This is the strongest transport-level security option, recommended for production environments.
• All three certificate files must be valid PEM-encoded files and must match each other.

Related

• Python SDK Reference
• TLS Setup