TLS Setup

Overview

Server-side TLS is the baseline transport security for any KubeMQ connection that leaves a trusted network — it encrypts the wire and lets the client confirm it's really talking to your KubeMQ server, not an impersonator. Reach for it whenever traffic crosses a public network or a boundary you don't fully control; skip it and channel names, payloads, and client IDs travel in plaintext with no protection against a spoofed endpoint.

It works by pairing the client with the CA certificate that signed the server's TLS certificate: TLSConfig(enabled=True, ca_file=...) loads that CA file, and the client performs a standard TLS handshake, validating the server's certificate chain before any request is sent. The client presents no certificate of its own — only the server proves its identity.

Gotchas: this is one-way trust — it stops eavesdropping and server impersonation, but the server still can't verify who the client is (that's what mTLS adds). ca_file must point to the issuing CA (or full chain), not the server's leaf certificate, or the handshake fails outright. And an expired or hostname-mismatched server certificate fails the same way as a missing CA path — read the raised error before assuming your CA file is the problem.

Prerequisites

• KubeMQ server running with TLS enabled
• Python SDK installed (pip install kubemq)
• TLS certificates (CA certificate file)

Code

"""Example: TLS configuration — connect to KubeMQ with TLS encryption."""

from __future__ import annotations

import asyncio

from kubemq import TLSConfig
from kubemq import AsyncPubSubClient, EventMessage


async def main() -> None:
    # TLS with server certificate verification (one-way TLS)
    tls_config = TLSConfig(
        enabled=True,
        ca_file="/path/to/ca.pem",
    )

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

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


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

How It Works

• TLSConfig with enabled=True and a ca_file path activates server-side TLS verification.
• The client verifies the server's certificate against the provided CA certificate before establishing the connection.
• Once connected, all traffic (including the event message) is encrypted in transit.
• Replace "/path/to/ca.pem" with the actual path to your CA certificate.

Related

• Python SDK Reference
• mTLS Setup