mirror of
https://github.com/home-assistant/core.git
synced 2026-09-27 01:46:11 -04:00
Fix K‑series that need encrypted pairing in SamsungTV (#180619)
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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/"
|
||||
}
|
||||
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user