From 07a04e73758d9ca1927c87e344b94ae8d4a55887 Mon Sep 17 00:00:00 2001 From: Simone Chemelli Date: Mon, 14 Sep 2026 09:24:29 +0200 Subject: [PATCH] =?UTF-8?q?Fix=20K=E2=80=91series=20that=20need=20encrypte?= =?UTF-8?q?d=20pairing=20in=20SamsungTV=20(#180619)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- homeassistant/components/samsungtv/bridge.py | 12 ++ .../components/samsungtv/config_flow.py | 28 +++- .../fixtures/device_info_UN55KU6290.json | 28 ++++ .../components/samsungtv/test_config_flow.py | 150 ++++++++++++++++++ 4 files changed, 217 insertions(+), 1 deletion(-) create mode 100644 tests/components/samsungtv/fixtures/device_info_UN55KU6290.json diff --git a/homeassistant/components/samsungtv/bridge.py b/homeassistant/components/samsungtv/bridge.py index 064f14abc418..e58723d8a219 100644 --- a/homeassistant/components/samsungtv/bridge.py +++ b/homeassistant/components/samsungtv/bridge.py @@ -98,6 +98,18 @@ def model_requires_encryption(model: str | None) -> bool: return model is not None and len(model) > 4 and model[4] in ("H", "J") +def model_may_require_encryption(model: str | None) -> bool: + """Return True for models that might need encrypted pairing. + + Some 2016 K-series models advertise the plain websocket method in their REST + device info but only pair via the encrypted PIN flow. Unlike H/J models (see + model_requires_encryption), other sets in the same series work with the + websocket method, so the encrypted method is only tried as a fallback once + websocket pairing has failed to connect. + """ + return model is not None and len(model) > 4 and model[4] == "K" + + async def async_get_device_info( hass: HomeAssistant, host: str, diff --git a/homeassistant/components/samsungtv/config_flow.py b/homeassistant/components/samsungtv/config_flow.py index 32b764d249f2..92b492d77777 100644 --- a/homeassistant/components/samsungtv/config_flow.py +++ b/homeassistant/components/samsungtv/config_flow.py @@ -39,7 +39,12 @@ from homeassistant.helpers.service_info.ssdp import ( ) from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo -from .bridge import SamsungTVBridge, async_get_device_info, mac_from_device_info +from .bridge import ( + SamsungTVBridge, + async_get_device_info, + mac_from_device_info, + model_may_require_encryption, +) from .const import ( CONF_MANUFACTURER, CONF_SESSION_ID, @@ -47,6 +52,7 @@ from .const import ( CONF_SSDP_RENDERING_CONTROL_LOCATION, DEFAULT_MANUFACTURER, DOMAIN, + ENCRYPTED_WEBSOCKET_PORT, LOGGER, METHOD_ENCRYPTED_WEBSOCKET, METHOD_LEGACY, @@ -324,6 +330,26 @@ class SamsungTVConfigFlow(ConfigFlow, domain=DOMAIN): result = await self._bridge.async_try_connect() if result == RESULT_SUCCESS: return self._get_entry_from_bridge() + if result == RESULT_CANNOT_CONNECT and model_may_require_encryption( + self._model + ): + # Some 2016 K-series sets advertise the websocket method but only + # pair via the encrypted PIN flow; try it before giving up + LOGGER.debug( + "Websocket pairing failed for %s (%s), falling back to encrypted", + self._host, + self._model, + ) + encrypted_bridge = SamsungTVBridge.get_bridge( + self.hass, + METHOD_ENCRYPTED_WEBSOCKET, + self._host, + ENCRYPTED_WEBSOCKET_PORT, + ) + if await encrypted_bridge.async_try_connect() == RESULT_CANNOT_CONNECT: + raise AbortFlow(RESULT_CANNOT_CONNECT) + self._bridge = encrypted_bridge + return await self.async_step_encrypted_pairing() if result != RESULT_AUTH_MISSING: raise AbortFlow(result) errors = {"base": RESULT_AUTH_MISSING} diff --git a/tests/components/samsungtv/fixtures/device_info_UN55KU6290.json b/tests/components/samsungtv/fixtures/device_info_UN55KU6290.json new file mode 100644 index 000000000000..f84a97ce14c2 --- /dev/null +++ b/tests/components/samsungtv/fixtures/device_info_UN55KU6290.json @@ -0,0 +1,28 @@ +{ + "id": "uuid:0dd7f9c9-b9a0-4b7e-8c3e-1f2b3c4d5e6f", + "name": "[TV] Samsung 6 Series (55)", + "version": "2.1.0", + "device": { + "type": "Samsung SmartTV", + "duid": "uuid:0dd7f9c9-b9a0-4b7e-8c3e-1f2b3c4d5e6f", + "model": "16_JAZZL_UHD_BASIC", + "modelName": "UN55KU6290", + "description": "Samsung DTV RCR", + "networkType": "wired", + "ssid": "", + "ip": "1.2.3.4", + "firmwareVersion": "Unknown", + "name": "[TV] Samsung 6 Series (55)", + "id": "uuid:0dd7f9c9-b9a0-4b7e-8c3e-1f2b3c4d5e6f", + "udn": "uuid:0dd7f9c9-b9a0-4b7e-8c3e-1f2b3c4d5e6f", + "resolution": "3840x2160", + "countryCode": "US", + "msfVersion": "2.1.0", + "smartHubAgreement": "true", + "wifiMac": "aa:bb:aa:aa:aa:aa", + "developerMode": "0", + "developerIP": "" + }, + "type": "Samsung SmartTV", + "uri": "https://1.2.3.4:8002/api/v2/" +} diff --git a/tests/components/samsungtv/test_config_flow.py b/tests/components/samsungtv/test_config_flow.py index 4c78e7e2762e..c1a3eba8736e 100644 --- a/tests/components/samsungtv/test_config_flow.py +++ b/tests/components/samsungtv/test_config_flow.py @@ -30,8 +30,11 @@ from homeassistant.components.samsungtv.const import ( CONF_SSDP_RENDERING_CONTROL_LOCATION, DEFAULT_MANUFACTURER, DOMAIN, + ENCRYPTED_WEBSOCKET_PORT, LEGACY_PORT, + METHOD_ENCRYPTED_WEBSOCKET, METHOD_LEGACY, + METHOD_WEBSOCKET, RESULT_AUTH_MISSING, RESULT_CANNOT_CONNECT, RESULT_NOT_SUPPORTED, @@ -283,6 +286,153 @@ async def test_user_encrypted_websocket( assert result4["result"].unique_id == "223da676-497a-4e06-9507-5e27ec4f0fb3" +@pytest.mark.usefixtures("rest_api", "remote_encrypted_websocket") +async def test_user_websocket_k_series_encrypted_fallback( + hass: HomeAssistant, rest_api: Mock +) -> None: + """Test a 2016 K-series set falls back to encrypted pairing (#177252). + + Its REST device info selects the websocket method, but the token handshake + times out (RESULT_CANNOT_CONNECT) because it only pairs via the encrypted + CloudPINPage flow. The encrypted port is reachable, so the flow probes it + successfully before committing to the encrypted pairing step. + """ + rest_api.rest_device_info.return_value = await async_load_json_object_fixture( + hass, "device_info_UN55KU6290.json", DOMAIN + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + + with ( + patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote.open", + side_effect=OSError("timed out"), + ), + patch( + "homeassistant.components.samsungtv.config_flow.SamsungTVEncryptedWSAsyncAuthenticator", + autospec=True, + ) as authenticator_mock, + ): + authenticator_mock.return_value.try_pin.side_effect = [ + None, + "037739871315caef138547b03e348b72", + ] + authenticator_mock.return_value.get_session_id_and_close.return_value = "1" + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_USER_DATA + ) + assert result2["type"] is FlowResultType.FORM + assert result2["step_id"] == "encrypted_pairing" + + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], user_input={CONF_PIN: "invalid"} + ) + assert result3["step_id"] == "encrypted_pairing" + assert result3["errors"] == {"base": "invalid_pin"} + + result4 = await hass.config_entries.flow.async_configure( + result3["flow_id"], user_input={CONF_PIN: "1234"} + ) + + assert result4["type"] is FlowResultType.CREATE_ENTRY + assert result4["data"][CONF_METHOD] == METHOD_ENCRYPTED_WEBSOCKET + assert result4["data"][CONF_MODEL] == "UN55KU6290" + assert result4["data"][CONF_PORT] == ENCRYPTED_WEBSOCKET_PORT + assert result4["data"][CONF_TOKEN] == "037739871315caef138547b03e348b72" + assert result4["data"][CONF_SESSION_ID] == "1" + + +@pytest.mark.usefixtures("remote_websocket", "rest_api") +async def test_user_websocket_k_series_stays_on_websocket( + hass: HomeAssistant, rest_api: Mock +) -> None: + """Test a K-series set that pairs over websocket is not pushed to encrypted. + + Regression guard for #70708 (UE32K5600): the encrypted fallback must only + trigger when the websocket pairing genuinely fails to connect. + """ + rest_api.rest_device_info.return_value = await async_load_json_object_fixture( + hass, "device_info_UN55KU6290.json", DOMAIN + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_USER_DATA + ) + + assert result2["type"] is FlowResultType.CREATE_ENTRY + assert result2["data"][CONF_METHOD] == METHOD_WEBSOCKET + assert result2["data"][CONF_MODEL] == "UN55KU6290" + assert result2["data"][CONF_PORT] == 8002 + + +@pytest.mark.usefixtures("rest_api") +async def test_user_websocket_non_k_series_cannot_connect( + hass: HomeAssistant, rest_api: Mock +) -> None: + """Test a non-K-series set that fails to connect is not pushed to encrypted. + + Regression guard for #70708: only K-series models get the encrypted + fallback, so a websocket connection failure on any other model must abort + with cannot_connect. + """ + rest_api.rest_device_info.return_value = await async_load_json_object_fixture( + hass, "device_info_UE43LS003.json", DOMAIN + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote.open", + side_effect=OSError("timed out"), + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_USER_DATA + ) + + assert result2["type"] is FlowResultType.ABORT + assert result2["reason"] == RESULT_CANNOT_CONNECT + + +@pytest.mark.usefixtures("rest_api", "remote_encrypted_websocket_failing") +async def test_user_websocket_k_series_encrypted_also_cannot_connect( + hass: HomeAssistant, rest_api: Mock +) -> None: + """Test a K-series set that is offline aborts cleanly with cannot_connect. + + Websocket pairing fails to connect, and the encrypted port is probed and + also unreachable (e.g. the TV is off), so the flow must abort with + cannot_connect instead of raising out of the encrypted pairing step. + """ + rest_api.rest_device_info.return_value = await async_load_json_object_fixture( + hass, "device_info_UN55KU6290.json", DOMAIN + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote.open", + side_effect=OSError("timed out"), + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_USER_DATA + ) + + assert result2["type"] is FlowResultType.ABORT + assert result2["reason"] == RESULT_CANNOT_CONNECT + + @pytest.mark.usefixtures("rest_api_failing") async def test_user_legacy_missing_auth(hass: HomeAssistant) -> None: """Test starting a flow by user with authentication."""