[mqtt] increase TLS options and compatibility (#14318)

Co-authored-by: J. Nick Koston <nick@home-assistant.io>
Co-authored-by: J. Nick Koston <nick@koston.org>

esphome/components/mqtt/__init__.py

@@ -1,8 +1,10 @@
 from esphome import automation
 import esphome.codegen as cg
 from esphome.components import logger, network, socket
+from esphome.components.const import CONF_VERIFY_SSL
 from esphome.components.esp32 import (
     add_idf_component,
+    allow_insecure_tls,
     idf_version,
     include_builtin_idf_component,
     request_tls,
@@ -79,6 +81,7 @@ def AUTO_LOAD():
     return ["json"]
 
 
+CONF_TLS = "tls"
 CONF_IDF_SEND_ASYNC = "idf_send_async"
 CONF_WAIT_FOR_CONNECTION = "wait_for_connection"
 CONF_TRANSPORT = "transport"
@@ -187,26 +190,8 @@ def _validate_transport(value: str) -> str:
 
 
 def validate_config(value):
-    transport = value.get(CONF_TRANSPORT)
-    # A certificate with plain ws is almost certainly a typo for wss
-    if transport == MQTT_TRANSPORT_WS and (
-        CONF_CERTIFICATE_AUTHORITY in value or CONF_CLIENT_CERTIFICATE in value
-    ):
-        raise cv.Invalid(
-            "transport 'ws' is unencrypted; remove the certificate options "
-            "or switch to transport 'wss'"
-        )
-    # TLS runs only with a CA or over wss, so a client certificate needs one of them
     if (
-        CONF_CLIENT_CERTIFICATE in value
-        and CONF_CERTIFICATE_AUTHORITY not in value
-        and transport != MQTT_TRANSPORT_WSS
-    ):
-        raise cv.Invalid(
-            "'client_certificate' requires 'certificate_authority' or transport 'wss'"
-        )
-    if (
-        transport not in (MQTT_TRANSPORT_WS, MQTT_TRANSPORT_WSS)
+        value.get(CONF_TRANSPORT) not in (MQTT_TRANSPORT_WS, MQTT_TRANSPORT_WSS)
         and CONF_WS_PATH in value
     ):
         raise cv.Invalid(
@@ -260,6 +245,47 @@ def validate_config(value):
     return out
 
 
+# Keys that only make sense over TLS; any of them switches it on
+TLS_IMPLYING_KEYS = frozenset(
+    {CONF_CERTIFICATE_AUTHORITY, CONF_VERIFY_SSL, CONF_CLIENT_CERTIFICATE}
+)
+
+
+def _derive_tls(config: ConfigType) -> ConfigType:
+    """Derive ``tls`` from the certificate options and the transport, reject contradictions, pick the port."""
+    # skip_cert_cn_check has a default on ESP32, so its value decides, not its presence
+    if config.get(CONF_VERIFY_SSL) is False and config.get(CONF_SKIP_CERT_CN_CHECK):
+        raise cv.Invalid(
+            "'skip_cert_cn_check' has no effect when 'verify_ssl' is false."
+        )
+    transport = config.get(CONF_TRANSPORT)
+    if (
+        config.keys() & TLS_IMPLYING_KEYS
+        or config.get(CONF_SKIP_CERT_CN_CHECK)
+        or transport == MQTT_TRANSPORT_WSS
+    ):
+        if config.get(CONF_TLS) is False:
+            raise cv.Invalid(
+                "'tls' cannot be false when a certificate, 'verify_ssl', "
+                "'skip_cert_cn_check' or transport 'wss' is set."
+            )
+        config[CONF_TLS] = True
+    # TLS with plain ws is almost certainly a typo for wss
+    if transport == MQTT_TRANSPORT_WS and config.get(CONF_TLS):
+        raise cv.Invalid(
+            "transport 'ws' is unencrypted; remove the TLS options "
+            "or switch to transport 'wss'"
+        )
+    config.setdefault(CONF_PORT, 8883 if config.get(CONF_TLS) else 1883)
+    return config
+
+
+_validate_tls = cv.All(
+    cv.only_when(CONF_CERTIFICATE_AUTHORITY, CONF_VERIFY_SSL, False, negate=True),
+    _derive_tls,
+)
+
+
 def _consume_mqtt_sockets(config: ConfigType) -> ConfigType:
     """Register socket needs for MQTT component."""
     # MQTT needs 1 socket for the broker connection
@@ -283,7 +309,7 @@ CONFIG_SCHEMA = cv.All(
             cv.GenerateID(): cv.declare_id(MQTTClientComponent),
             cv.Required(CONF_BROKER): cv.string_strict,
             cv.Optional(CONF_ENABLE_ON_BOOT, default=True): cv.boolean,
-            cv.Optional(CONF_PORT, default=1883): cv.port,
+            cv.Optional(CONF_PORT): cv.port,
             cv.Optional(CONF_USERNAME, default=""): cv.string,
             cv.Optional(CONF_PASSWORD, default=""): cv.sensitive(),
             cv.Optional(CONF_CLEAN_SESSION, default=False): cv.boolean,
@@ -295,6 +321,8 @@ CONFIG_SCHEMA = cv.All(
             cv.SplitDefault(CONF_IDF_SEND_ASYNC, esp32=False): cv.All(
                 cv.boolean, cv.only_on_esp32
             ),
+            cv.Optional(CONF_TLS): cv.All(cv.boolean, cv.only_on_esp32),
+            cv.Optional(CONF_VERIFY_SSL): cv.All(cv.boolean, cv.only_on_esp32),
             cv.Optional(CONF_CERTIFICATE_AUTHORITY): cv.All(
                 cv.string, cv.only_on_esp32
             ),
@@ -356,6 +384,7 @@ CONFIG_SCHEMA = cv.All(
         }
     ),
     validate_config,
+    _validate_tls,
     cv.only_on(
         [
             PLATFORM_BK72XX,
@@ -497,21 +526,28 @@ async def to_code(config):
 
     # esp-idf only
     transport = config.get(CONF_TRANSPORT)
-    ca = cg.add_optional(var.set_ca_certificate, config.get(CONF_CERTIFICATE_AUTHORITY))
-    if ca is not None or transport == MQTT_TRANSPORT_WSS:
-        cg.add(var.set_skip_cert_cn_check(config[CONF_SKIP_CERT_CN_CHECK]))
+    if config.get(CONF_TLS):
+        cg.add_define("USE_MQTT_TLS")
         if (client_certificate := config.get(CONF_CLIENT_CERTIFICATE)) is not None:
             cg.add(var.set_cl_certificate(client_certificate))
             cg.add(var.set_cl_key(config[CONF_CLIENT_CERTIFICATE_KEY]))
+        if config.get(CONF_SKIP_CERT_CN_CHECK):
+            cg.add(var.set_skip_cert_cn_check(True))
+        # Trust source: none, the given CA, or the certificate bundle
+        if config.get(CONF_VERIFY_SSL) is False:
+            cg.add_define("USE_MQTT_TLS_INSECURE")
+            allow_insecure_tls()
+        elif (ca := config.get(CONF_CERTIFICATE_AUTHORITY)) is not None:
+            cg.add(var.set_ca_certificate(ca))
+        else:
+            cg.add_define("USE_MQTT_CERT_BUNDLE")
+            require_certificate_bundle()
 
     if CORE.is_esp32:
         if transport != MQTT_TRANSPORT_TCP:
             cg.add(var.set_transport(transport))
         if transport in (MQTT_TRANSPORT_WS, MQTT_TRANSPORT_WSS):
             cg.add(var.set_ws_path(config.get(CONF_WS_PATH, DEFAULT_WS_PATH)))
-        # wss without an explicit CA verifies the broker against the bundled roots
-        if transport == MQTT_TRANSPORT_WSS and ca is None:
-            require_certificate_bundle()
 
     if CONF_IDF_SEND_ASYNC in config and config[CONF_IDF_SEND_ASYNC]:
         cg.add_define("USE_MQTT_IDF_ENQUEUE")

esphome/components/mqtt/mqtt_backend_esp32.cpp

@@ -8,7 +8,7 @@
 #include "esphome/core/log.h"
 #include "esphome/core/application.h"
 
-#if CONFIG_MBEDTLS_CERTIFICATE_BUNDLE
+#ifdef USE_MQTT_CERT_BUNDLE
 #include "esp_crt_bundle.h"
 #endif
 
@@ -43,46 +43,36 @@ bool MQTTBackendESP32::initialize_() {
   if (!this->client_id_.empty()) {
     mqtt_cfg_.credentials.client_id = this->client_id_.c_str();
   }
-
-  // Plain TCP keeps upgrading to SSL when a CA certificate is configured
-  bool uses_tls = false;
+#ifdef USE_MQTT_TLS
+  mqtt_cfg_.broker.verification.skip_cert_common_name_check = this->skip_cert_cn_check_;
+  mqtt_cfg_.broker.verification.certificate = this->ca_certificate_;
+#ifdef USE_MQTT_CERT_BUNDLE
+  mqtt_cfg_.broker.verification.crt_bundle_attach = esp_crt_bundle_attach;
+#endif
+  // Validation only accepts the client certificate together with its key
+  mqtt_cfg_.credentials.authentication.certificate = this->cl_certificate_;
+  mqtt_cfg_.credentials.authentication.key = this->cl_key_;
+#endif
+  // Validation keeps TLS off plain ws, so only the TCP transport has two forms
   switch (this->transport_) {
     case MQTTTransport::MQTT_TRANSPORT_WS:
       mqtt_cfg_.broker.address.transport = MQTT_TRANSPORT_OVER_WS;
       break;
     case MQTTTransport::MQTT_TRANSPORT_WSS:
       mqtt_cfg_.broker.address.transport = MQTT_TRANSPORT_OVER_WSS;
-      uses_tls = true;
       break;
     case MQTTTransport::MQTT_TRANSPORT_TCP:
     default:
-      if (this->ca_certificate_.has_value()) {
-        mqtt_cfg_.broker.address.transport = MQTT_TRANSPORT_OVER_SSL;
-        uses_tls = true;
-      } else {
-        mqtt_cfg_.broker.address.transport = MQTT_TRANSPORT_OVER_TCP;
-      }
+#ifdef USE_MQTT_TLS
+      mqtt_cfg_.broker.address.transport = MQTT_TRANSPORT_OVER_SSL;
+#else
+      mqtt_cfg_.broker.address.transport = MQTT_TRANSPORT_OVER_TCP;
+#endif
       break;
   }
   // Only set for the WebSocket transports
   mqtt_cfg_.broker.address.path = this->ws_path_;
 
-  if (uses_tls) {
-    if (this->ca_certificate_.has_value()) {
-      mqtt_cfg_.broker.verification.certificate = this->ca_certificate_.value().c_str();
-#if CONFIG_MBEDTLS_CERTIFICATE_BUNDLE
-    } else {
-      // wss without a CA verifies the broker against the bundled roots, as http_request does
-      mqtt_cfg_.broker.verification.crt_bundle_attach = esp_crt_bundle_attach;
-#endif
-    }
-    mqtt_cfg_.broker.verification.skip_cert_common_name_check = this->skip_cert_cn_check_;
-    if (this->cl_certificate_.has_value() && this->cl_key_.has_value()) {
-      mqtt_cfg_.credentials.authentication.certificate = this->cl_certificate_.value().c_str();
-      mqtt_cfg_.credentials.authentication.key = this->cl_key_.value().c_str();
-    }
-  }
-
   auto *mqtt_client = esp_mqtt_client_init(&mqtt_cfg_);
   if (mqtt_client) {
     handler_.reset(mqtt_client);
@@ -91,7 +81,11 @@ bool MQTTBackendESP32::initialize_() {
 #if defined(USE_MQTT_IDF_ENQUEUE)
     // Create the task only after MQTT client is initialized successfully
     // Use larger stack size when TLS is enabled
-    size_t stack_size = uses_tls ? TASK_STACK_SIZE_TLS : TASK_STACK_SIZE;
+#ifdef USE_MQTT_TLS
+    constexpr size_t stack_size = TASK_STACK_SIZE_TLS;
+#else
+    constexpr size_t stack_size = TASK_STACK_SIZE;
+#endif
     xTaskCreate(esphome_mqtt_task, "esphome_mqtt", stack_size, (void *) this, TASK_PRIORITY, &this->task_handle_);
     if (this->task_handle_ == nullptr) {
       ESP_LOGE(TAG, "Failed to create MQTT task");

esphome/components/mqtt/mqtt_backend_esp32.h

@@ -214,10 +214,13 @@ class MQTTBackendESP32 final : public MQTTBackend {
 
   void loop() final;
 
-  void set_ca_certificate(const std::string &cert) { ca_certificate_ = cert; }
-  void set_cl_certificate(const std::string &cert) { cl_certificate_ = cert; }
-  void set_cl_key(const std::string &key) { cl_key_ = key; }
-  void set_skip_cert_cn_check(bool skip_check) { skip_cert_cn_check_ = skip_check; }
+#ifdef USE_MQTT_TLS
+  // The certificates are string literals from code generation, so only the pointers are kept
+  void set_ca_certificate(const char *cert) { this->ca_certificate_ = cert; }
+  void set_cl_certificate(const char *cert) { this->cl_certificate_ = cert; }
+  void set_cl_key(const char *key) { this->cl_key_ = key; }
+  void set_skip_cert_cn_check(bool skip_check) { this->skip_cert_cn_check_ = skip_check; }
+#endif
   void set_transport(MQTTTransport transport) { this->transport_ = transport; }
   void set_ws_path(const char *path) { this->ws_path_ = path; }
   MQTTTransport get_transport() const { return this->transport_; }
@@ -256,10 +259,12 @@ class MQTTBackendESP32 final : public MQTTBackend {
   std::string client_id_;
   uint16_t keep_alive_;
   bool clean_session_;
-  optional<std::string> ca_certificate_;
-  optional<std::string> cl_certificate_;
-  optional<std::string> cl_key_;
+#ifdef USE_MQTT_TLS
+  const char *ca_certificate_{nullptr};
+  const char *cl_certificate_{nullptr};
+  const char *cl_key_{nullptr};
   bool skip_cert_cn_check_{false};
+#endif
   MQTTTransport transport_{MQTTTransport::MQTT_TRANSPORT_TCP};
   const char *ws_path_{nullptr};
 #if defined(USE_MQTT_IDF_ENQUEUE)

esphome/components/mqtt/mqtt_client.cpp

@@ -222,6 +222,13 @@ void MQTTClientComponent::dump_config() {
                   transport == MQTTTransport::MQTT_TRANSPORT_WSS ? LOG_STR_LITERAL("wss") : LOG_STR_LITERAL("ws"),
                   this->mqtt_backend_.get_ws_path());
   }
+#endif
+#if defined(USE_MQTT_CERT_BUNDLE)
+  ESP_LOGCONFIG(TAG, "  TLS: certificate bundle");
+#elif defined(USE_MQTT_TLS_INSECURE)
+  ESP_LOGCONFIG(TAG, "  TLS: no verification");
+#elif defined(USE_MQTT_TLS)
+  ESP_LOGCONFIG(TAG, "  TLS: CA certificate");
 #endif
   if (this->is_discovery_ip_enabled()) {
     ESP_LOGCONFIG(TAG, "  Discovery IP enabled");

esphome/components/mqtt/mqtt_client.h

@@ -147,11 +147,13 @@ class MQTTClientComponent final : public Component {
   bool is_discovery_enabled() const;
   bool is_discovery_ip_enabled() const;
 
-#ifdef USE_ESP32
+#if defined(USE_ESP32) && defined(USE_MQTT_TLS)
   void set_ca_certificate(const char *cert) { this->mqtt_backend_.set_ca_certificate(cert); }
   void set_cl_certificate(const char *cert) { this->mqtt_backend_.set_cl_certificate(cert); }
   void set_cl_key(const char *key) { this->mqtt_backend_.set_cl_key(key); }
   void set_skip_cert_cn_check(bool skip_check) { this->mqtt_backend_.set_skip_cert_cn_check(skip_check); }
+#endif
+#ifdef USE_ESP32
   void set_transport(MQTTTransport transport) { this->mqtt_backend_.set_transport(transport); }
   void set_ws_path(const char *path) { this->mqtt_backend_.set_ws_path(path); }
 #endif

esphome/core/defines.h

@@ -434,6 +434,9 @@
 #define USE_ESP32_CRASH_HANDLER
 #define USE_ESP32_VASPRINTF_STUB
 #define USE_MQTT_IDF_ENQUEUE
+#define USE_MQTT_TLS
+#define USE_MQTT_CERT_BUNDLE
+#define USE_MQTT_TLS_INSECURE
 #define USE_ESPHOME_TASK_LOG_BUFFER
 #define ESPHOME_TASK_LOG_BUFFER_SIZE 768
 #define USE_OTA_ROLLBACK

esphome/mqtt.py

@@ -97,11 +97,12 @@ def prepare(
 
     # Match the firmware's transport; imported here to keep component code off the CLI start-up path
     from esphome.components.mqtt import (
+        CONF_TLS,
         CONF_TRANSPORT,
+        CONF_VERIFY_SSL,
         CONF_WS_PATH,
         DEFAULT_WS_PATH,
         MQTT_TRANSPORT_TCP,
-        MQTT_TRANSPORT_WSS,
     )
 
     transport = config[CONF_MQTT].get(CONF_TRANSPORT, MQTT_TRANSPORT_TCP)
@@ -122,15 +123,21 @@ def prepare(
     elif username:
         client.username_pw_set(username, password)
 
-    # TLS with the CA, or the system CAs over wss
-    certificate_authority = config[CONF_MQTT].get(CONF_CERTIFICATE_AUTHORITY)
-    if certificate_authority or transport == MQTT_TRANSPORT_WSS:
-        context = ssl.create_default_context(cadata=certificate_authority)
-        if config[CONF_MQTT].get(CONF_SKIP_CERT_CN_CHECK):
+    mqtt_conf = config[CONF_MQTT]
+    # Validation sets tls whenever a certificate option or the wss transport implies it
+    if mqtt_conf.get(CONF_TLS):
+        if mqtt_conf.get(CONF_VERIFY_SSL) is False:
+            context = ssl.create_default_context()
             context.check_hostname = False
-        if config[CONF_MQTT].get(CONF_CLIENT_CERTIFICATE) and config[CONF_MQTT].get(
-            CONF_CLIENT_CERTIFICATE_KEY
-        ):
+            context.verify_mode = ssl.CERT_NONE
+        else:
+            # Without a CA the system trust store stands in for the firmware's bundle
+            context = ssl.create_default_context(
+                cadata=mqtt_conf.get(CONF_CERTIFICATE_AUTHORITY)
+            )
+            if mqtt_conf.get(CONF_SKIP_CERT_CN_CHECK):
+                context.check_hostname = False
+        if mqtt_conf.get(CONF_CLIENT_CERTIFICATE):
             with (
                 tempfile.NamedTemporaryFile(
                     encoding="utf-8", mode="w+", delete=False
@@ -140,8 +147,8 @@ def prepare(
                 ) as key_file,
             ):
                 try:
-                    cert_file.write(config[CONF_MQTT].get(CONF_CLIENT_CERTIFICATE))
-                    key_file.write(config[CONF_MQTT].get(CONF_CLIENT_CERTIFICATE_KEY))
+                    cert_file.write(mqtt_conf[CONF_CLIENT_CERTIFICATE])
+                    key_file.write(mqtt_conf[CONF_CLIENT_CERTIFICATE_KEY])
                     cert_file.close()
                     key_file.close()
                     context.load_cert_chain(cert_file.name, key_file.name)

tests/component_tests/mqtt/test_transport_validation.py

@@ -10,6 +10,7 @@ from esphome.components.mqtt import (
     MQTT_TRANSPORT_TCP,
     MQTT_TRANSPORT_WS,
     MQTT_TRANSPORT_WSS,
+    _validate_tls,
     _validate_transport,
     validate_config,
 )
@@ -86,22 +87,11 @@ def test_plaintext_ws_with_certificate_rejected(
 ) -> None:
     """Test that pairing a CA or client certificate with unencrypted ws is rejected."""
     with pytest.raises(cv.Invalid, match="unencrypted"):
-        validate_config(
-            {
-                CONF_TRANSPORT: MQTT_TRANSPORT_WS,
-                CONF_TOPIC_PREFIX: "test",
-                **certificate_options,
-            }
-        )
+        _validate_tls({CONF_TRANSPORT: MQTT_TRANSPORT_WS, **certificate_options})
 
 
-def test_client_certificate_requires_tls() -> None:
-    """Test that a client certificate without a CA is only accepted over wss."""
-    certificate = {
-        CONF_CLIENT_CERTIFICATE: "-----BEGIN CERTIFICATE-----",
-        CONF_CLIENT_CERTIFICATE_KEY: "-----BEGIN PRIVATE KEY-----",
-        CONF_TOPIC_PREFIX: "test",
-    }
-    with pytest.raises(cv.Invalid, match="requires 'certificate_authority'"):
-        validate_config({CONF_TRANSPORT: MQTT_TRANSPORT_TCP, **certificate})
-    validate_config({CONF_TRANSPORT: MQTT_TRANSPORT_WSS, **certificate})
+def test_wss_switches_tls_on() -> None:
+    """Test that wss implies tls and cannot be combined with tls false."""
+    assert _validate_tls({CONF_TRANSPORT: MQTT_TRANSPORT_WSS})["tls"] is True
+    with pytest.raises(cv.Invalid, match="'tls' cannot be false"):
+        _validate_tls({CONF_TRANSPORT: MQTT_TRANSPORT_WSS, "tls": False})

tests/component_tests/mqtt/test_validation.py

@@ -0,0 +1,72 @@
+"""Tests for the mqtt TLS validation rules and the port default."""
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.const import CONF_VERIFY_SSL
+from esphome.components.mqtt import CONF_TLS, _validate_tls
+from esphome.const import (
+    CONF_BROKER,
+    CONF_CERTIFICATE_AUTHORITY,
+    CONF_CLIENT_CERTIFICATE,
+    CONF_PORT,
+    CONF_SKIP_CERT_CN_CHECK,
+    CONF_TOPIC_PREFIX,
+)
+from esphome.types import ConfigType
+
+BASE: ConfigType = {CONF_BROKER: "broker.local", CONF_TOPIC_PREFIX: "node"}
+
+
+def _validate(**extra: object) -> ConfigType:
+    return _validate_tls({**BASE, **extra})
+
+
+def test_plain_defaults_to_port_1883() -> None:
+    out = _validate()
+    assert out[CONF_PORT] == 1883
+    assert CONF_TLS not in out
+
+
+@pytest.mark.parametrize(
+    "extra",
+    [
+        {CONF_TLS: True},
+        {CONF_CERTIFICATE_AUTHORITY: "cert"},
+        {CONF_VERIFY_SSL: False},
+        {CONF_CLIENT_CERTIFICATE: "cert"},
+        {CONF_SKIP_CERT_CN_CHECK: True},
+    ],
+)
+def test_tls_defaults_to_port_8883(extra: ConfigType) -> None:
+    out = _validate(**extra)
+    assert out[CONF_TLS] is True
+    assert out[CONF_PORT] == 8883
+
+
+def test_explicit_port_is_kept() -> None:
+    assert _validate(**{CONF_TLS: True, CONF_PORT: 1234})[CONF_PORT] == 1234
+
+
+def test_certificate_authority_with_verify_ssl_false_rejected() -> None:
+    with pytest.raises(cv.Invalid, match="only valid when 'verify_ssl' is not false"):
+        _validate(**{CONF_CERTIFICATE_AUTHORITY: "cert", CONF_VERIFY_SSL: False})
+
+
+def test_skip_cert_cn_check_with_verify_ssl_false_rejected() -> None:
+    with pytest.raises(cv.Invalid, match="'skip_cert_cn_check' has no effect"):
+        _validate(**{CONF_SKIP_CERT_CN_CHECK: True, CONF_VERIFY_SSL: False})
+
+
+@pytest.mark.parametrize(
+    "extra",
+    [
+        {CONF_CERTIFICATE_AUTHORITY: "cert"},
+        {CONF_VERIFY_SSL: True},
+        {CONF_CLIENT_CERTIFICATE: "cert"},
+        {CONF_SKIP_CERT_CN_CHECK: True},
+    ],
+)
+def test_tls_false_with_certificate_options_rejected(extra: ConfigType) -> None:
+    with pytest.raises(cv.Invalid, match="'tls' cannot be false"):
+        _validate(**{CONF_TLS: False, **extra})

tests/components/mqtt/test.esp32-idf.yaml

@@ -1,3 +1,6 @@
 packages:
   common: !include common.yaml
   update: !include common-update.yaml
+
+mqtt:
+  tls: true

tests/components/mqtt/validate-tls-ca.esp32-idf.yaml

@@ -0,0 +1,10 @@
+packages:
+  common: !include common.yaml
+
+mqtt:
+  broker: test.mosquitto.org
+  skip_cert_cn_check: true
+  certificate_authority: |
+    -----BEGIN CERTIFICATE-----
+    MIIBszCCAVoCCQDAAAAAAAAAADAKBggqhkjOPQQDAjAfMR0wGwYDVQQDDBRtcXR0
+    -----END CERTIFICATE-----

tests/components/mqtt/validate-tls-insecure.esp32-idf.yaml

@@ -0,0 +1,6 @@
+packages:
+  common: !include common.yaml
+
+mqtt:
+  broker: test.mosquitto.org
+  verify_ssl: false

tests/unit_tests/test_mqtt.py

@@ -3,21 +3,29 @@
 from __future__ import annotations
 
 import json
+from pathlib import Path
+import ssl
 import threading
 import time
 from unittest.mock import MagicMock, patch
 
 import pytest
 
+from esphome.components.const import CONF_VERIFY_SSL
+from esphome.components.mqtt import CONF_TLS
 from esphome.const import (
     CONF_BIRTH_MESSAGE,
     CONF_BROKER,
+    CONF_CERTIFICATE_AUTHORITY,
+    CONF_CLIENT_CERTIFICATE,
+    CONF_CLIENT_CERTIFICATE_KEY,
     CONF_ESPHOME,
     CONF_MQTT,
     CONF_NAME,
     CONF_PAYLOAD,
     CONF_PORT,
     CONF_RETAIN,
+    CONF_SKIP_CERT_CN_CHECK,
     CONF_TOPIC,
 )
 from esphome.core import EsphomeError
@@ -483,6 +491,9 @@ def test_prepare_follows_the_firmware_transport(
         config[CONF_MQTT]["transport"] = transport
     if ws_path is not None:
         config[CONF_MQTT]["ws_path"] = ws_path
+    if expect_tls:
+        # Validation switches tls on for wss
+        config[CONF_MQTT][CONF_TLS] = True
     client = mock_client_class.return_value
 
     assert prepare(config, [], None, None, None, None, None) is client
@@ -500,3 +511,85 @@ def test_prepare_follows_the_firmware_transport(
     else:
         client.tls_set_context.assert_not_called()
     client.connect.assert_called_once_with("mqtt.local", 1883)
+
+
+def _mqtt_config(options: dict | None = None) -> dict:
+    return {
+        CONF_MQTT: {CONF_BROKER: "broker.local", CONF_PORT: 8883, **(options or {})}
+    }
+
+
+def _prepare_with_mocks(config: dict) -> tuple[MagicMock, MagicMock, MagicMock]:
+    """Runs prepare() against a mocked paho client and SSL context factory."""
+    with (
+        patch("esphome.mqtt.mqtt.Client") as client_cls,
+        patch("esphome.mqtt.ssl.create_default_context") as create_context,
+    ):
+        client = client_cls.return_value
+        context = create_context.return_value
+        # A real context verifies hostnames unless prepare() turns that off
+        context.check_hostname = True
+        prepare(config, [], None, None, None, None, None)
+    return client, create_context, context
+
+
+def test_prepare_without_tls_connects_in_plaintext() -> None:
+    client, create_context, _ = _prepare_with_mocks(_mqtt_config({CONF_PORT: 1883}))
+
+    create_context.assert_not_called()
+    client.tls_set_context.assert_not_called()
+    client.connect.assert_called_once_with("broker.local", 1883)
+
+
+def test_prepare_tls_without_ca_uses_system_trust_store() -> None:
+    client, create_context, context = _prepare_with_mocks(
+        _mqtt_config({CONF_TLS: True})
+    )
+
+    create_context.assert_called_once_with(cadata=None)
+    assert context.check_hostname is True
+    client.tls_set_context.assert_called_once_with(context)
+
+
+def test_prepare_tls_with_ca_and_skip_cn_check() -> None:
+    client, create_context, context = _prepare_with_mocks(
+        _mqtt_config(
+            {
+                CONF_TLS: True,
+                CONF_CERTIFICATE_AUTHORITY: "---cert---",
+                CONF_SKIP_CERT_CN_CHECK: True,
+            }
+        )
+    )
+
+    create_context.assert_called_once_with(cadata="---cert---")
+    assert context.check_hostname is False
+    client.tls_set_context.assert_called_once_with(context)
+
+
+def test_prepare_verify_ssl_false_disables_verification() -> None:
+    client, create_context, context = _prepare_with_mocks(
+        _mqtt_config({CONF_TLS: True, CONF_VERIFY_SSL: False})
+    )
+
+    create_context.assert_called_once_with()
+    assert context.check_hostname is False
+    assert context.verify_mode == ssl.CERT_NONE
+    client.tls_set_context.assert_called_once_with(context)
+
+
+def test_prepare_loads_client_certificate() -> None:
+    _, _, context = _prepare_with_mocks(
+        _mqtt_config(
+            {
+                CONF_TLS: True,
+                CONF_CLIENT_CERTIFICATE: "---client cert---",
+                CONF_CLIENT_CERTIFICATE_KEY: "---client key---",
+            }
+        )
+    )
+
+    context.load_cert_chain.assert_called_once()
+    cert_file, key_file = context.load_cert_chain.call_args.args
+    assert not Path(cert_file).exists()
+    assert not Path(key_file).exists()