Add MQTT token pairing to Victron GX (#180590)

homeassistant/components/victron_gx/config_flow.py

@@ -5,10 +5,22 @@ import logging
 from typing import Any, override
 from urllib.parse import urlparse
 
-from victron_mqtt import AuthenticationError, CannotConnectError, Hub as VictronVenusHub
+from victron_mqtt import (
+    AuthenticationError,
+    CannotConnectError,
+    Hub as VictronVenusHub,
+    PairingError,
+    PairingToken,
+    request_pairing_token,
+)
 import voluptuous as vol
 
-from homeassistant.config_entries import SOURCE_IGNORE, ConfigFlow, ConfigFlowResult
+from homeassistant.config_entries import (
+    SOURCE_IGNORE,
+    SOURCE_REAUTH,
+    ConfigFlow,
+    ConfigFlowResult,
+)
 from homeassistant.const import (
     CONF_HOST,
     CONF_MODEL,
@@ -21,10 +33,11 @@ from homeassistant.helpers import selector
 from homeassistant.helpers.redact import async_redact_data
 from homeassistant.helpers.service_info.ssdp import SsdpServiceInfo
 
-from .const import CONF_INSTALLATION_ID, CONF_SERIAL, DOMAIN
+from .const import CONF_INSTALLATION_ID, CONF_MQTT_TOKEN_PAIRING, CONF_SERIAL, DOMAIN
 
 DEFAULT_HOST = "venus.local"
 DEFAULT_PORT = 1883
+DEFAULT_SSL_PORT = 8883
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -46,11 +59,10 @@ STEP_USER_DATA_SCHEMA = vol.Schema(
 
 STEP_SSDP_AUTH_DATA_SCHEMA = vol.Schema(
     {
-        vol.Optional(CONF_USERNAME, default=""): selector.TextSelector(),
         vol.Optional(CONF_PASSWORD, default=""): selector.TextSelector(
             selector.TextSelectorConfig(type=selector.TextSelectorType.PASSWORD)
         ),
-        vol.Optional(CONF_SSL, default=False): selector.BooleanSelector(),
+        vol.Optional(CONF_SSL): selector.BooleanSelector(),
     }
 )
 
@@ -110,6 +122,8 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN):
         self.installation_id: str | None = None
         self.friendly_name: str | None = None
         self.model_name: str | None = None
+        self.mqtt_token_pairing = False
+        self.ssdp_use_ssl = True
 
     @override
     async def async_step_user(
@@ -177,6 +191,7 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN):
         self.installation_id = discovery_info.upnp["X_VrmPortalId"]
         self.model_name = discovery_info.upnp["modelName"]
         self.friendly_name = discovery_info.upnp["friendlyName"]
+        self.mqtt_token_pairing = discovery_info.upnp.get("X_MqttTokenPairing") == "1"
 
         await self.async_set_unique_id(self.installation_id)
 
@@ -207,25 +222,6 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN):
             "name": self.friendly_name or self.hostname
         }
 
-        # Verify connectivity before showing the confirmation dialog
-        try:
-            ssdp_conf = {
-                CONF_HOST: self.hostname,
-                CONF_PORT: DEFAULT_PORT,
-                CONF_SERIAL: self.serial,
-                CONF_INSTALLATION_ID: self.installation_id,
-            }
-            await validate_input(ssdp_conf)
-        except AuthenticationError:
-            return await self.async_step_ssdp_auth()
-        except CannotConnectError:
-            return self.async_abort(reason="cannot_connect")
-        except Exception:
-            _LOGGER.exception(
-                "Unexpected error validating SSDP discovery for Victron GX"
-            )
-            return self.async_abort(reason="unknown")
-
         return await self.async_step_ssdp_confirm()
 
     async def async_step_ssdp_confirm(
@@ -236,19 +232,52 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN):
         assert self.installation_id is not None
 
         if user_input is not None:
+            data: dict[str, Any] = {
+                CONF_HOST: self.hostname,
+                CONF_PORT: DEFAULT_SSL_PORT,
+                CONF_SERIAL: self.serial,
+                CONF_INSTALLATION_ID: self.installation_id,
+                CONF_MODEL: self.model_name,
+                CONF_SSL: True,
+            }
+            try:
+                await validate_input(data)
+            except AuthenticationError:
+                if self.mqtt_token_pairing:
+                    return await self.async_step_ssdp_token_pairing()
+                return await self.async_step_ssdp_auth()
+            except CannotConnectError:
+                data[CONF_PORT] = DEFAULT_PORT
+                data[CONF_SSL] = False
+                self.ssdp_use_ssl = False
+                try:
+                    await validate_input(data)
+                except AuthenticationError:
+                    if self.mqtt_token_pairing:
+                        return await self.async_step_ssdp_token_pairing()
+                    return await self.async_step_ssdp_auth()
+                except CannotConnectError:
+                    if self.mqtt_token_pairing:
+                        return await self.async_step_ssdp_token_pairing()
+                    return self.async_abort(reason="cannot_connect")
+                except Exception:
+                    _LOGGER.exception(
+                        "Unexpected error validating SSDP discovery for Victron GX"
+                    )
+                    return self.async_abort(reason="unknown")
+            except Exception:
+                _LOGGER.exception(
+                    "Unexpected error validating SSDP discovery for Victron GX"
+                )
+                return self.async_abort(reason="unknown")
+
             return self.async_create_entry(
                 title=ENTRY_TITLE_FORMAT.format(
                     installation_id=self.installation_id,
                     host=self.hostname,
-                    port=DEFAULT_PORT,
+                    port=data[CONF_PORT],
                 ),
-                data={
-                    CONF_HOST: self.hostname,
-                    CONF_PORT: DEFAULT_PORT,
-                    CONF_SERIAL: self.serial,
-                    CONF_INSTALLATION_ID: self.installation_id,
-                    CONF_MODEL: self.model_name,
-                },
+                data=data,
             )
 
         self._set_confirm_only()
@@ -257,6 +286,66 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN):
             description_placeholders={"name": self.friendly_name or self.hostname},
         )
 
+    async def async_step_ssdp_token_pairing(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle automatic token pairing with the GX device."""
+        assert self.hostname is not None
+        assert self.installation_id is not None
+
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            try:
+                credentials: PairingToken = await request_pairing_token(
+                    self.hostname, self.installation_id
+                )
+            except PairingError:
+                errors["base"] = "pairing_failed"
+            except Exception:
+                _LOGGER.exception("Failed to connect to GX device for token pairing")
+                errors["base"] = "cannot_connect"
+            else:
+                data: dict[str, Any] = {
+                    CONF_HOST: self.hostname,
+                    CONF_PORT: DEFAULT_SSL_PORT,
+                    CONF_SERIAL: self.serial,
+                    CONF_INSTALLATION_ID: self.installation_id,
+                    CONF_MODEL: self.model_name,
+                    CONF_MQTT_TOKEN_PAIRING: True,
+                    CONF_USERNAME: credentials.token_name,
+                    CONF_PASSWORD: credentials.password,
+                    CONF_SSL: True,
+                }
+                try:
+                    await validate_input(data)
+                except AuthenticationError:
+                    errors["base"] = "invalid_auth"
+                except CannotConnectError:
+                    errors["base"] = "cannot_connect"
+                except Exception:
+                    _LOGGER.exception("Unexpected error validating paired credentials")
+                    errors["base"] = "unknown"
+                else:
+                    if self.source == SOURCE_REAUTH:
+                        return self.async_update_reload_and_abort(
+                            self._get_reauth_entry(), data_updates=data
+                        )
+                    return self.async_create_entry(
+                        title=ENTRY_TITLE_FORMAT.format(
+                            installation_id=self.installation_id,
+                            host=self.hostname,
+                            port=DEFAULT_SSL_PORT,
+                        ),
+                        data=data,
+                    )
+
+        self._set_confirm_only()
+        return self.async_show_form(
+            step_id="ssdp_token_pairing",
+            errors=errors,
+            description_placeholders={CONF_HOST: self.hostname},
+        )
+
     async def async_step_ssdp_auth(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
@@ -271,14 +360,16 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN):
                 "SSDP auth user input received: %s",
                 async_redact_data(user_input, TO_REDACT),
             )
+            use_ssl = user_input.get(CONF_SSL, self.ssdp_use_ssl)
             data: dict[str, Any] = {
                 CONF_HOST: self.hostname,
-                CONF_PORT: DEFAULT_PORT,
+                CONF_PORT: DEFAULT_SSL_PORT if use_ssl else DEFAULT_PORT,
                 CONF_SERIAL: self.serial,
                 CONF_INSTALLATION_ID: self.installation_id,
-                CONF_USERNAME: user_input.get(CONF_USERNAME),
-                CONF_PASSWORD: user_input.get(CONF_PASSWORD),
-                CONF_SSL: user_input.get(CONF_SSL),
+                CONF_MODEL: self.model_name,
+                CONF_USERNAME: "remoteconsole",
+                CONF_PASSWORD: user_input.get(CONF_PASSWORD) or None,
+                CONF_SSL: use_ssl,
             }
 
             try:
@@ -298,7 +389,7 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN):
                     title=ENTRY_TITLE_FORMAT.format(
                         installation_id=self.installation_id,
                         host=self.hostname,
-                        port=DEFAULT_PORT,
+                        port=data[CONF_PORT],
                     ),
                     data=data,
                 )
@@ -306,7 +397,8 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN):
         return self.async_show_form(
             step_id="ssdp_auth",
             data_schema=self.add_suggested_values_to_schema(
-                STEP_SSDP_AUTH_DATA_SCHEMA, user_input
+                STEP_SSDP_AUTH_DATA_SCHEMA,
+                user_input or {CONF_SSL: self.ssdp_use_ssl},
             ),
             errors=errors,
             description_placeholders={CONF_HOST: self.hostname},
@@ -368,6 +460,13 @@ class VictronGXConfigFlow(ConfigFlow, domain=DOMAIN):
 
     async def async_step_reauth(self, _: Mapping[str, Any]) -> ConfigFlowResult:
         """Handle reauthentication."""
+        reauth_entry = self._get_reauth_entry()
+        if reauth_entry.data.get(CONF_MQTT_TOKEN_PAIRING):
+            self.hostname = reauth_entry.data[CONF_HOST]
+            self.serial = reauth_entry.data.get(CONF_SERIAL)
+            self.installation_id = reauth_entry.data[CONF_INSTALLATION_ID]
+            self.model_name = reauth_entry.data.get(CONF_MODEL)
+            return await self.async_step_ssdp_token_pairing()
         return await self.async_step_reauth_confirm()
 
     async def async_step_reauth_confirm(

homeassistant/components/victron_gx/const.py

@@ -3,6 +3,7 @@
 DOMAIN = "victron_gx"
 
 CONF_INSTALLATION_ID = "installation_id"
+CONF_MQTT_TOKEN_PAIRING = "mqtt_token_pairing"
 CONF_SERIAL = "serial"
 
 # Binary sensor enum ids must be "on" for on and "off" for off.

homeassistant/components/victron_gx/manifest.json

@@ -12,6 +12,10 @@
     {
       "X_MqttOnLan": "1",
       "manufacturer": "Victron Energy"
+    },
+    {
+      "X_MqttTokenPairing": "1",
+      "manufacturer": "Victron Energy"
     }
   ]
 }

homeassistant/components/victron_gx/strings.json

@@ -118,6 +118,7 @@
     "error": {
       "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
       "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "pairing_failed": "Token pairing failed. Ensure pairing mode is active on the GX device and try again.",
       "unknown": "[%key:common::config_flow::error::unknown%]"
     },
     "step": {
@@ -153,21 +154,23 @@
       },
       "ssdp_auth": {
         "data": {
-          "password": "[%key:common::config_flow::data::password%]",
-          "ssl": "[%key:common::config_flow::data::ssl%]",
-          "username": "[%key:common::config_flow::data::username%]"
+          "password": "GX password",
+          "ssl": "[%key:common::config_flow::data::ssl%]"
         },
         "data_description": {
           "password": "[%key:component::victron_gx::config::step::user::data_description::password%]",
-          "ssl": "[%key:component::victron_gx::config::step::user::data_description::ssl%]",
-          "username": "[%key:component::victron_gx::config::step::user::data_description::username%]"
+          "ssl": "[%key:component::victron_gx::config::step::user::data_description::ssl%]"
         },
-        "description": "Authentication is required to connect to {host}.",
+        "description": "Enter the GX password for the GX device at {host}.",
         "title": "Authenticate Victron GX"
       },
       "ssdp_confirm": {
         "description": "Do you want to set up the Victron GX device {name}?"
       },
+      "ssdp_token_pairing": {
+        "description": "Enable pairing mode on the GX device before submitting:\n\n- Via the GX user interface: **Settings** > **Integrations** > **MQTT Devices** > **Pairing mode**\n- On GX devices without a built-in screen: Quickly double-press the built-in button\n\nPairing mode is active for 120 seconds.",
+        "title": "Pair with GX device"
+      },
       "user": {
         "data": {
           "host": "[%key:common::config_flow::data::host%]",

homeassistant/generated/ssdp.py

@@ -389,6 +389,10 @@ SSDP = {
             "X_MqttOnLan": "1",
             "manufacturer": "Victron Energy",
         },
+        {
+            "X_MqttTokenPairing": "1",
+            "manufacturer": "Victron Energy",
+        },
     ],
     "webostv": [
         {

tests/components/victron_gx/test_config_flow.py

@@ -3,11 +3,20 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from victron_mqtt import AuthenticationError, CannotConnectError
+from victron_mqtt import (
+    AuthenticationError,
+    CannotConnectError,
+    PairingError,
+    PairingToken,
+)
 
-from homeassistant.components.victron_gx.config_flow import DEFAULT_PORT
+from homeassistant.components.victron_gx.config_flow import (
+    DEFAULT_PORT,
+    DEFAULT_SSL_PORT,
+)
 from homeassistant.components.victron_gx.const import (
     CONF_INSTALLATION_ID,
+    CONF_MQTT_TOKEN_PAIRING,
     CONF_SERIAL,
     DOMAIN,
 )
@@ -47,6 +56,25 @@ def assert_entry_title(
     assert result["title"] == f"Victron OS {installation_id} ({host}:{port})"
 
 
+def ssdp_discovery_info(token_pairing: str = "0") -> SsdpServiceInfo:
+    """Return Victron GX SSDP discovery data."""
+    upnp = {
+        "serialNumber": MOCK_SERIAL,
+        "X_VrmPortalId": MOCK_INSTALLATION_ID,
+        "modelName": MOCK_MODEL,
+        "friendlyName": MOCK_FRIENDLY_NAME,
+        "X_MqttOnLan": "1",
+        "X_MqttTokenPairing": token_pairing,
+        "manufacturer": "Victron Energy",
+    }
+    return SsdpServiceInfo(
+        ssdp_usn="mock_usn",
+        ssdp_st="upnp:rootdevice",
+        ssdp_location="http://192.168.1.100:80/",
+        upnp=upnp,
+    )
+
+
 @pytest.fixture
 def mock_victron_hub():
     """Mock the Victron Hub."""
@@ -234,13 +262,14 @@ async def test_ssdp_flow_success(hass: HomeAssistant) -> None:
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
     assert result["result"].unique_id == MOCK_INSTALLATION_ID
-    assert_entry_title(result)
+    assert_entry_title(result, port=DEFAULT_SSL_PORT)
     assert result["data"] == {
         CONF_HOST: MOCK_HOST,
-        CONF_PORT: DEFAULT_PORT,
+        CONF_PORT: DEFAULT_SSL_PORT,
         CONF_SERIAL: MOCK_SERIAL,
         CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID,
         CONF_MODEL: MOCK_MODEL,
+        CONF_SSL: True,
     }
 
 
@@ -280,6 +309,13 @@ async def test_ssdp_discovery_error(
         data=discovery_info,
     )
 
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "ssdp_confirm"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+
     assert result["type"] is FlowResultType.ABORT
     assert result["reason"] == reason
 
@@ -342,6 +378,13 @@ async def test_ssdp_flow_auth_required(
         data=discovery_info,
     )
 
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "ssdp_confirm"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "ssdp_auth"
 
@@ -352,22 +395,22 @@ async def test_ssdp_flow_auth_required(
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         user_input={
-            CONF_USERNAME: "test-username",
             CONF_PASSWORD: "test-password",
         },
     )
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
     assert result["result"].unique_id == MOCK_INSTALLATION_ID
-    assert_entry_title(result)
+    assert_entry_title(result, port=DEFAULT_SSL_PORT)
     assert result["data"] == {
         CONF_HOST: MOCK_HOST,
-        CONF_PORT: DEFAULT_PORT,
+        CONF_PORT: DEFAULT_SSL_PORT,
         CONF_SERIAL: MOCK_SERIAL,
         CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID,
-        CONF_USERNAME: "test-username",
+        CONF_MODEL: MOCK_MODEL,
+        CONF_USERNAME: "remoteconsole",
         CONF_PASSWORD: "test-password",
-        CONF_SSL: False,
+        CONF_SSL: True,
     }
 
 
@@ -399,6 +442,13 @@ async def test_ssdp_auth_invalid_credentials(
         data=discovery_info,
     )
 
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "ssdp_confirm"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "ssdp_auth"
 
@@ -410,8 +460,8 @@ async def test_ssdp_auth_invalid_credentials(
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         user_input={
-            CONF_USERNAME: "wrong-user",
             CONF_PASSWORD: "wrong-password",
+            CONF_SSL: False,
         },
     )
 
@@ -425,8 +475,8 @@ async def test_ssdp_auth_invalid_credentials(
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         user_input={
-            CONF_USERNAME: "test-user",
             CONF_PASSWORD: "test-password",
+            CONF_SSL: False,
         },
     )
 
@@ -438,7 +488,8 @@ async def test_ssdp_auth_invalid_credentials(
         CONF_PORT: DEFAULT_PORT,
         CONF_SERIAL: MOCK_SERIAL,
         CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID,
-        CONF_USERNAME: "test-user",
+        CONF_MODEL: MOCK_MODEL,
+        CONF_USERNAME: "remoteconsole",
         CONF_PASSWORD: "test-password",
         CONF_SSL: False,
     }
@@ -482,6 +533,13 @@ async def test_ssdp_auth_error(
         data=discovery_info,
     )
 
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "ssdp_confirm"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "ssdp_auth"
 
@@ -490,8 +548,8 @@ async def test_ssdp_auth_error(
     result = await hass.config_entries.flow.async_configure(
         result["flow_id"],
         user_input={
-            CONF_USERNAME: "test-user",
             CONF_PASSWORD: "test-password",
+            CONF_SSL: False,
         },
     )
 
@@ -641,6 +699,48 @@ async def test_reauth_flow_clears_credentials(
     assert mock_config_entry.data[CONF_SSL] is False
 
 
+async def test_reauth_flow_renews_pairing_token(
+    hass: HomeAssistant, mock_victron_hub: MagicMock
+) -> None:
+    """Test reauthentication renews generated pairing credentials."""
+    token_entry = MockConfigEntry(
+        domain=DOMAIN,
+        unique_id=MOCK_INSTALLATION_ID,
+        data={
+            CONF_HOST: MOCK_HOST,
+            CONF_PORT: DEFAULT_SSL_PORT,
+            CONF_USERNAME: "token/homeassistant/old",
+            CONF_PASSWORD: "old-password",
+            CONF_SSL: True,
+            CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID,
+            CONF_MQTT_TOKEN_PAIRING: True,
+            CONF_MODEL: MOCK_MODEL,
+            CONF_SERIAL: MOCK_SERIAL,
+        },
+        title=(f"Victron OS {MOCK_INSTALLATION_ID} ({MOCK_HOST}:{DEFAULT_SSL_PORT})"),
+    )
+    token_entry.add_to_hass(hass)
+
+    result = await token_entry.start_reauth_flow(hass)
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "ssdp_token_pairing"
+
+    with patch(
+        "homeassistant.components.victron_gx.config_flow.request_pairing_token",
+        return_value=PairingToken("token/homeassistant/new", "new-generated-password"),
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={}
+        )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reauth_successful"
+    assert token_entry.data[CONF_USERNAME] == "token/homeassistant/new"
+    assert token_entry.data[CONF_PASSWORD] == "new-generated-password"
+    assert token_entry.data[CONF_MQTT_TOKEN_PAIRING] is True
+    mock_victron_hub.return_value.connect.assert_awaited_once()
+
+
 @pytest.mark.parametrize(
     ("exception", "error"),
     [
@@ -923,3 +1023,270 @@ async def test_ssdp_flow_abort_on_invalid_hostname(
 
     assert result["type"] is FlowResultType.ABORT
     assert result["reason"] == "cannot_connect"
+
+
+async def test_ssdp_flow_falls_back_to_plain_mqtt(
+    hass: HomeAssistant, mock_victron_hub: MagicMock
+) -> None:
+    """Test SSDP setup falls back to plain MQTT when TLS is unavailable."""
+    mock_victron_hub.return_value.connect.side_effect = [
+        CannotConnectError("TLS unavailable"),
+        None,
+    ]
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_SSDP},
+        data=ssdp_discovery_info(),
+    )
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert_entry_title(result)
+    assert result["data"][CONF_PORT] == DEFAULT_PORT
+    assert result["data"][CONF_SSL] is False
+
+
+@pytest.mark.parametrize(
+    ("second_exception", "result_type", "result_key", "result_value"),
+    [
+        pytest.param(
+            AuthenticationError("Authentication required"),
+            FlowResultType.FORM,
+            "step_id",
+            "ssdp_auth",
+            id="authentication-required",
+        ),
+        pytest.param(
+            Exception("Unexpected error"),
+            FlowResultType.ABORT,
+            "reason",
+            "unknown",
+            id="unexpected-error",
+        ),
+    ],
+)
+async def test_ssdp_flow_tls_fallback_error(
+    hass: HomeAssistant,
+    mock_victron_hub: MagicMock,
+    second_exception: Exception,
+    result_type: FlowResultType,
+    result_key: str,
+    result_value: str,
+) -> None:
+    """Test errors while falling back from TLS during SSDP setup."""
+    mock_victron_hub.return_value.connect.side_effect = [
+        CannotConnectError("TLS unavailable"),
+        second_exception,
+    ]
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_SSDP},
+        data=ssdp_discovery_info(),
+    )
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+
+    assert result["type"] is result_type
+    assert result[result_key] == result_value
+
+
+async def test_ssdp_flow_preserves_plain_mqtt_for_authentication(
+    hass: HomeAssistant, mock_victron_hub: MagicMock
+) -> None:
+    """Test SSDP authentication preserves the discovered plaintext transport."""
+    mock_victron_hub.return_value.connect.side_effect = [
+        CannotConnectError("TLS unavailable"),
+        AuthenticationError("Authentication required"),
+        None,
+    ]
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_SSDP},
+        data=ssdp_discovery_info(),
+    )
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "ssdp_auth"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={CONF_PASSWORD: "test-password"}
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert_entry_title(result)
+    assert result["data"][CONF_PORT] == DEFAULT_PORT
+    assert result["data"][CONF_SSL] is False
+
+
+@pytest.mark.parametrize(
+    "initial_exceptions",
+    [
+        pytest.param(
+            [AuthenticationError("Authentication required"), None],
+            id="authentication-required",
+        ),
+        pytest.param(
+            [
+                CannotConnectError("TLS unavailable"),
+                AuthenticationError("Authentication required"),
+                None,
+            ],
+            id="plain-mqtt-authentication-required",
+        ),
+        pytest.param(
+            [
+                CannotConnectError("TLS unavailable"),
+                CannotConnectError("Plain MQTT unavailable"),
+                None,
+            ],
+            id="mqtt-unavailable",
+        ),
+    ],
+)
+async def test_ssdp_token_pairing_success(
+    hass: HomeAssistant,
+    mock_victron_hub: MagicMock,
+    initial_exceptions: list[Exception | None],
+) -> None:
+    """Test SSDP token pairing creates an entry with generated credentials."""
+    mock_victron_hub.return_value.connect.side_effect = initial_exceptions
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_SSDP},
+        data=ssdp_discovery_info("1"),
+    )
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "ssdp_token_pairing"
+
+    with patch(
+        "homeassistant.components.victron_gx.config_flow.request_pairing_token",
+        return_value=PairingToken(
+            token_name="token/homeassistant/homeassistant_abc123",
+            password="generated-password",
+        ),
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={}
+        )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert_entry_title(result, port=DEFAULT_SSL_PORT)
+    assert result["data"] == {
+        CONF_HOST: MOCK_HOST,
+        CONF_PORT: DEFAULT_SSL_PORT,
+        CONF_SERIAL: MOCK_SERIAL,
+        CONF_INSTALLATION_ID: MOCK_INSTALLATION_ID,
+        CONF_MODEL: MOCK_MODEL,
+        CONF_MQTT_TOKEN_PAIRING: True,
+        CONF_USERNAME: "token/homeassistant/homeassistant_abc123",
+        CONF_PASSWORD: "generated-password",
+        CONF_SSL: True,
+    }
+
+
+@pytest.mark.parametrize(
+    ("exception", "error"),
+    [
+        pytest.param(
+            PairingError("Pairing mode inactive"),
+            "pairing_failed",
+            id="pairing-rejected",
+        ),
+        pytest.param(
+            ConnectionError("Connection refused"),
+            "cannot_connect",
+            id="connection-error",
+        ),
+    ],
+)
+async def test_ssdp_token_pairing_request_error(
+    hass: HomeAssistant,
+    mock_victron_hub: MagicMock,
+    exception: Exception,
+    error: str,
+) -> None:
+    """Test errors while requesting token pairing credentials."""
+    mock_victron_hub.return_value.connect.side_effect = AuthenticationError(
+        "Authentication required"
+    )
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_SSDP},
+        data=ssdp_discovery_info("1"),
+    )
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+
+    with patch(
+        "homeassistant.components.victron_gx.config_flow.request_pairing_token",
+        side_effect=exception,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={}
+        )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "ssdp_token_pairing"
+    assert result["errors"] == {"base": error}
+
+
+@pytest.mark.parametrize(
+    ("exception", "error"),
+    [
+        pytest.param(
+            AuthenticationError("Invalid credentials"),
+            "invalid_auth",
+            id="invalid-auth",
+        ),
+        pytest.param(
+            CannotConnectError("Cannot connect"),
+            "cannot_connect",
+            id="cannot-connect",
+        ),
+        pytest.param(Exception("Unexpected error"), "unknown", id="unknown"),
+    ],
+)
+async def test_ssdp_token_pairing_validation_error(
+    hass: HomeAssistant,
+    mock_victron_hub: MagicMock,
+    exception: Exception,
+    error: str,
+) -> None:
+    """Test errors while validating generated pairing credentials."""
+    mock_victron_hub.return_value.connect.side_effect = [
+        AuthenticationError("Authentication required"),
+        exception,
+    ]
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_SSDP},
+        data=ssdp_discovery_info("1"),
+    )
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input={}
+    )
+
+    with patch(
+        "homeassistant.components.victron_gx.config_flow.request_pairing_token",
+        return_value=PairingToken("token/homeassistant/test", "password"),
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={}
+        )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "ssdp_token_pairing"
+    assert result["errors"] == {"base": error}