jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-22 20:57:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add PSK auth and SSDP discovery to Bravia TV (#77772)

Browse Source
This commit is contained in:
Artem Draft 2022-09-23 16:03:43 +03:00 committed by GitHub
parent 83b426deb0
commit 7c460cc641
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 302 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
homeassistant/components/braviatv/__init__.py
View File

@ -11,7 +11,7 @@ from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import CONF_IGNORED_SOURCES, DOMAIN
from .const import CONF_IGNORED_SOURCES, CONF_USE_PSK, DOMAIN
from .coordinator import BraviaTVCoordinator
PLATFORMS: Final[list[Platform]] = [Platform.MEDIA_PLAYER, Platform.REMOTE]
@ -22,13 +22,20 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
host = config_entry.data[CONF_HOST]
mac = config_entry.data[CONF_MAC]
pin = config_entry.data[CONF_PIN]
use_psk = config_entry.data.get(CONF_USE_PSK, False)
ignored_sources = config_entry.options.get(CONF_IGNORED_SOURCES, [])
session = async_create_clientsession(
hass, cookie_jar=CookieJar(unsafe=True, quote_cookie=False)
)
client = BraviaTV(host, mac, session=session)
coordinator = BraviaTVCoordinator(hass, client, pin, ignored_sources)
coordinator = BraviaTVCoordinator(
hass=hass,
client=client,
pin=pin,
use_psk=use_psk,
ignored_sources=ignored_sources,
)
config_entry.async_on_unload(config_entry.add_update_listener(update_listener))
await coordinator.async_config_entry_first_refresh()

96
homeassistant/components/braviatv/config_flow.py
View File

@ -2,14 +2,16 @@
from __future__ import annotations
from typing import Any
from urllib.parse import urlparse
from aiohttp import CookieJar
from pybravia import BraviaTV, BraviaTVError, BraviaTVNotSupported
from pybravia import BraviaTV, BraviaTVAuthError, BraviaTVError, BraviaTVNotSupported
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, CONF_PIN
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_create_clientsession
@ -23,6 +25,7 @@ from .const import (
ATTR_MODEL,
CLIENTID_PREFIX,
CONF_IGNORED_SOURCES,
CONF_USE_PSK,
DOMAIN,
NICKNAME,
)
@ -33,10 +36,9 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
client: BraviaTV
def __init__(self) -> None:
"""Initialize config flow."""
self.client: BraviaTV | None = None
self.device_config: dict[str, Any] = {}
@staticmethod
@ -45,11 +47,28 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Bravia TV options callback."""
return BraviaTVOptionsFlowHandler(config_entry)
async def async_init_device(self) -> FlowResult:
"""Initialize and create Bravia TV device from config."""
pin = self.device_config[CONF_PIN]
def create_client(self) -> None:
"""Create Bravia TV client from config."""
host = self.device_config[CONF_HOST]
session = async_create_clientsession(
self.hass,
cookie_jar=CookieJar(unsafe=True, quote_cookie=False),
)
self.client = BraviaTV(host=host, session=session)
await self.client.connect(pin=pin, clientid=CLIENTID_PREFIX, nickname=NICKNAME)
async def async_create_device(self) -> FlowResult:
"""Initialize and create Bravia TV device from config."""
assert self.client
pin = self.device_config[CONF_PIN]
use_psk = self.device_config[CONF_USE_PSK]
if use_psk:
await self.client.connect(psk=pin)
else:
await self.client.connect(
pin=pin, clientid=CLIENTID_PREFIX, nickname=NICKNAME
)
await self.client.set_wol_mode(True)
system_info = await self.client.get_system_info()
@ -72,13 +91,8 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is not None:
host = user_input[CONF_HOST]
if is_host_valid(host):
session = async_create_clientsession(
self.hass,
cookie_jar=CookieJar(unsafe=True, quote_cookie=False),
)
self.client = BraviaTV(host=host, session=session)
self.device_config[CONF_HOST] = host
self.create_client()
return await self.async_step_authorize()
errors[CONF_HOST] = "invalid_host"
@ -92,18 +106,23 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_authorize(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Get PIN from the Bravia TV device."""
"""Authorize Bravia TV device."""
errors: dict[str, str] = {}
if user_input is not None:
self.device_config[CONF_PIN] = user_input[CONF_PIN]
self.device_config[CONF_USE_PSK] = user_input[CONF_USE_PSK]
try:
return await self.async_init_device()
return await self.async_create_device()
except BraviaTVAuthError:
errors["base"] = "invalid_auth"
except BraviaTVNotSupported:
errors["base"] = "unsupported_model"
except BraviaTVError:
errors["base"] = "cannot_connect"
assert self.client
try:
await self.client.pair(CLIENTID_PREFIX, NICKNAME)
except BraviaTVError:
@ -111,10 +130,53 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form(
step_id="authorize",
data_schema=vol.Schema({vol.Required(CONF_PIN, default=""): str}),
data_schema=vol.Schema(
{
vol.Required(CONF_PIN, default=""): str,
vol.Required(CONF_USE_PSK, default=False): bool,
}
),
errors=errors,
)
async def async_step_ssdp(self, discovery_info: ssdp.SsdpServiceInfo) -> FlowResult:
"""Handle a discovered device."""
parsed_url = urlparse(discovery_info.ssdp_location)
host = parsed_url.hostname
await self.async_set_unique_id(discovery_info.upnp[ssdp.ATTR_UPNP_UDN])
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
self._async_abort_entries_match({CONF_HOST: host})
scalarweb_info = discovery_info.upnp["X_ScalarWebAPI_DeviceInfo"]
service_types = scalarweb_info["X_ScalarWebAPI_ServiceList"][
"X_ScalarWebAPI_ServiceType"
]
if "videoScreen" not in service_types:
return self.async_abort(reason="not_bravia_device")
model_name = discovery_info.upnp[ssdp.ATTR_UPNP_MODEL_NAME]
friendly_name = discovery_info.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME]
self.context["title_placeholders"] = {
CONF_NAME: f"{model_name} ({friendly_name})",
CONF_HOST: host,
}
self.device_config[CONF_HOST] = host
return await self.async_step_confirm()
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Allow the user to confirm adding the device."""
if user_input is not None:
self.create_client()
return await self.async_step_authorize()
return self.async_show_form(step_id="confirm")
class BraviaTVOptionsFlowHandler(config_entries.OptionsFlow):
"""Config flow options for Bravia TV."""

1
homeassistant/components/braviatv/const.py
View File

@ -9,6 +9,7 @@ ATTR_MANUFACTURER: Final = "Sony"
ATTR_MODEL: Final = "model"
CONF_IGNORED_SOURCES: Final = "ignored_sources"
CONF_USE_PSK: Final = "use_psk"
CLIENTID_PREFIX: Final = "HomeAssistant"
DOMAIN: Final = "braviatv"

11
homeassistant/components/braviatv/coordinator.py
View File

@ -59,12 +59,14 @@ class BraviaTVCoordinator(DataUpdateCoordinator[None]):
hass: HomeAssistant,
client: BraviaTV,
pin: str,
use_psk: bool,
ignored_sources: list[str],
) -> None:
"""Initialize Bravia TV Client."""
self.client = client
self.pin = pin
self.use_psk = use_psk
self.ignored_sources = ignored_sources
self.source: str | None = None
self.source_list: list[str] = []
@ -110,9 +112,12 @@ class BraviaTVCoordinator(DataUpdateCoordinator[None]):
"""Connect and fetch data."""
try:
if not self.connected:
await self.client.connect(
pin=self.pin, clientid=CLIENTID_PREFIX, nickname=NICKNAME
)
if self.use_psk:
await self.client.connect(psk=self.pin)
else:
await self.client.connect(
pin=self.pin, clientid=CLIENTID_PREFIX, nickname=NICKNAME
)
self.connected = True
power_status = await self.client.get_power_status()

6
homeassistant/components/braviatv/manifest.json
View File

@ -4,6 +4,12 @@
"documentation": "https://www.home-assistant.io/integrations/braviatv",
"requirements": ["pybravia==0.2.2"],
"codeowners": ["@bieniu", "@Drafteed"],
"ssdp": [
{
"st": "urn:schemas-sony-com:service:ScalarWebAPI:1",
"manufacturer": "Sony Corporation"
}
],
"config_flow": true,
"iot_class": "local_polling",
"loggers": ["pybravia"]

12
homeassistant/components/braviatv/strings.json
View File

@ -9,20 +9,26 @@
},
"authorize": {
"title": "Authorize Sony Bravia TV",
"description": "Enter the PIN code shown on the Sony Bravia TV. \n\nIf the PIN code is not shown, you have to unregister Home Assistant on your TV, go to: Settings -> Network -> Remote device settings -> Unregister remote device.",
"description": "Enter the PIN code shown on the Sony Bravia TV. \n\nIf the PIN code is not shown, you have to unregister Home Assistant on your TV, go to: Settings -> Network -> Remote device settings -> Deregister remote device. \n\nYou can use PSK (Pre-Shared-Key) instead of PIN. PSK is a user-defined secret key used for access control. This authentication method is recommended as more stable. To enable PSK on your TV, go to: Settings -> Network -> Home Network Setup -> IP Control. Then check «Use PSK authentication» box and enter your PSK instead of PIN.",
"data": {
"pin": "[%key:common::config_flow::data::pin%]"
"pin": "[%key:common::config_flow::data::pin%]",
"use_psk": "Use PSK authentication"
}
},
"confirm": {
"description": "[%key:common::config_flow::description::confirm_setup%]"
}
},
"error": {
"invalid_host": "[%key:common::config_flow::error::invalid_host%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unsupported_model": "Your TV model is not supported."
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"no_ip_control": "IP Control is disabled on your TV or the TV is not supported."
"no_ip_control": "IP Control is disabled on your TV or the TV is not supported.",
"not_bravia_device": "The device is not a Bravia TV."
}
},
"options": {

12
homeassistant/components/braviatv/translations/en.json
View File

@ -2,21 +2,27 @@
"config": {
"abort": {
"already_configured": "Device is already configured",
"no_ip_control": "IP Control is disabled on your TV or the TV is not supported."
"no_ip_control": "IP Control is disabled on your TV or the TV is not supported.",
"not_bravia_device": "The device is not a Bravia TV."
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"invalid_host": "Invalid hostname or IP address",
"unsupported_model": "Your TV model is not supported."
},
"step": {
"authorize": {
"data": {
"pin": "PIN Code"
"pin": "PIN Code",
"use_psk": "Use PSK authentication"
},
"description": "Enter the PIN code shown on the Sony Bravia TV. \n\nIf the PIN code is not shown, you have to unregister Home Assistant on your TV, go to: Settings -> Network -> Remote device settings -> Unregister remote device.",
"description": "Enter the PIN code shown on the Sony Bravia TV. \n\nIf the PIN code is not shown, you have to unregister Home Assistant on your TV, go to: Settings -> Network -> Remote device settings -> Deregister remote device. \n\nYou can use PSK (Pre-Shared-Key) instead of PIN. PSK is a user-defined secret key used for access control. This authentication method is recommended as more stable. To enable PSK on your TV, go to: Settings -> Network -> Home Network Setup -> IP Control. Then check \u00abUse PSK authentication\u00bb box and enter your PSK instead of PIN.",
"title": "Authorize Sony Bravia TV"
},
"confirm": {
"description": "Do you want to start set up?"
},
"user": {
"data": {
"host": "Host"

6
homeassistant/generated/ssdp.py
View File

@ -15,6 +15,12 @@ SSDP = {
"manufacturer": "AXIS",
},
],
"braviatv": [
{
"manufacturer": "Sony Corporation",
"st": "urn:schemas-sony-com:service:ScalarWebAPI:1"
}
],
"control4": [
{
"st": "c4:director",

183
tests/components/braviatv/test_config_flow.py
View File

@ -1,11 +1,16 @@
"""Define tests for the Bravia TV config flow."""
from unittest.mock import patch
from pybravia import BraviaTVConnectionError, BraviaTVNotSupported
from pybravia import BraviaTVAuthError, BraviaTVConnectionError, BraviaTVNotSupported
from homeassistant import data_entry_flow
from homeassistant.components.braviatv.const import CONF_IGNORED_SOURCES, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.components import ssdp
from homeassistant.components.braviatv.const import (
CONF_IGNORED_SOURCES,
CONF_USE_PSK,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN
from tests.common import MockConfigEntry
@ -31,6 +36,44 @@ BRAVIA_SOURCES = [
{"title": "AV/Component", "uri": "extInput:component?port=1"},
]
BRAVIA_SSDP = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://bravia-host:52323/dmr.xml",
upnp={
ssdp.ATTR_UPNP_UDN: "uuid:1234",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Living TV",
ssdp.ATTR_UPNP_MODEL_NAME: "KE-55XH9096",
"X_ScalarWebAPI_DeviceInfo": {
"X_ScalarWebAPI_ServiceList": {
"X_ScalarWebAPI_ServiceType": [
"guide",
"system",
"audio",
"avContent",
"videoScreen",
],
},
},
},
)
FAKE_BRAVIA_SSDP = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://soundbar-host:52323/dmr.xml",
upnp={
ssdp.ATTR_UPNP_UDN: "uuid:1234",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Sony Audio Device",
ssdp.ATTR_UPNP_MODEL_NAME: "HT-S700RF",
"X_ScalarWebAPI_DeviceInfo": {
"X_ScalarWebAPI_ServiceList": {
"X_ScalarWebAPI_ServiceType": ["guide", "system", "audio", "avContent"],
},
},
},
)
async def test_show_form(hass):
"""Test that the form is served with no input."""
@ -42,6 +85,83 @@ async def test_show_form(hass):
assert result["step_id"] == SOURCE_USER
async def test_ssdp_discovery(hass):
"""Test that the device is discovered."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data=BRAVIA_SSDP,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "confirm"
with patch("pybravia.BraviaTV.connect"), patch("pybravia.BraviaTV.pair"), patch(
"pybravia.BraviaTV.set_wol_mode"
), patch(
"pybravia.BraviaTV.get_system_info",
return_value=BRAVIA_SYSTEM_INFO,
), patch(
"homeassistant.components.braviatv.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "authorize"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "1234", CONF_USE_PSK: False}
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == "very_unique_string"
assert result["title"] == "TV-Model"
assert result["data"] == {
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_USE_PSK: False,
CONF_MAC: "AA:BB:CC:DD:EE:FF",
}
async def test_ssdp_discovery_fake(hass):
"""Test that not Bravia device is not discovered."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data=FAKE_BRAVIA_SSDP,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "not_bravia_device"
async def test_ssdp_discovery_exist(hass):
"""Test that the existed device is not discovered."""
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id="very_unique_string",
data={
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
},
title="TV-Model",
)
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data=BRAVIA_SSDP,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_user_invalid_host(hass):
"""Test that errors are shown when the host is invalid."""
result = await hass.config_entries.flow.async_init(
@ -51,6 +171,22 @@ async def test_user_invalid_host(hass):
assert result["errors"] == {CONF_HOST: "invalid_host"}
async def test_authorize_invalid_auth(hass):
"""Test that authorization errors shown on the authorization step."""
with patch(
"pybravia.BraviaTV.connect",
side_effect=BraviaTVAuthError,
), patch("pybravia.BraviaTV.pair"):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "1234"}
)
assert result["errors"] == {"base": "invalid_auth"}
async def test_authorize_cannot_connect(hass):
"""Test that errors are shown when cannot connect to host at the authorize step."""
with patch(
@ -114,7 +250,6 @@ async def test_duplicate_error(hass):
"pybravia.BraviaTV.get_system_info",
return_value=BRAVIA_SYSTEM_INFO,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
)
@ -136,7 +271,6 @@ async def test_create_entry(hass):
), patch(
"homeassistant.components.braviatv.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
)
@ -145,7 +279,7 @@ async def test_create_entry(hass):
assert result["step_id"] == "authorize"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "1234"}
result["flow_id"], user_input={CONF_PIN: "1234", CONF_USE_PSK: False}
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
@ -154,6 +288,7 @@ async def test_create_entry(hass):
assert result["data"] == {
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_USE_PSK: False,
CONF_MAC: "AA:BB:CC:DD:EE:FF",
}
@ -168,7 +303,6 @@ async def test_create_entry_with_ipv6_address(hass):
), patch(
"homeassistant.components.braviatv.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
@ -179,7 +313,7 @@ async def test_create_entry_with_ipv6_address(hass):
assert result["step_id"] == "authorize"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "1234"}
result["flow_id"], user_input={CONF_PIN: "1234", CONF_USE_PSK: False}
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
@ -188,6 +322,39 @@ async def test_create_entry_with_ipv6_address(hass):
assert result["data"] == {
CONF_HOST: "2001:db8::1428:57ab",
CONF_PIN: "1234",
CONF_USE_PSK: False,
CONF_MAC: "AA:BB:CC:DD:EE:FF",
}
async def test_create_entry_psk(hass):
"""Test that the user step works with PSK auth."""
with patch("pybravia.BraviaTV.connect"), patch("pybravia.BraviaTV.pair"), patch(
"pybravia.BraviaTV.set_wol_mode"
), patch(
"pybravia.BraviaTV.get_system_info",
return_value=BRAVIA_SYSTEM_INFO,
), patch(
"homeassistant.components.braviatv.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "authorize"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "mypsk", CONF_USE_PSK: True}
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == "very_unique_string"
assert result["title"] == "TV-Model"
assert result["data"] == {
CONF_HOST: "bravia-host",
CONF_PIN: "mypsk",
CONF_USE_PSK: True,
CONF_MAC: "AA:BB:CC:DD:EE:FF",
}