jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Blink auth flow improvement and mini camera support (#38027)

Browse Source
This commit is contained in:
Kevin Fronczak 2020-08-05 06:21:14 -04:00 committed by GitHub
parent 3fc5f9deb8
commit 3fdec7946c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 271 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
homeassistant/components/blink/__init__.py
View File

@ -1,32 +1,25 @@
"""Support for Blink Home Camera System."""
import asyncio
from copy import deepcopy
import logging
from blinkpy.auth import Auth
from blinkpy.blinkpy import Blink
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_FILENAME,
CONF_NAME,
CONF_PASSWORD,
CONF_PIN,
CONF_SCAN_INTERVAL,
CONF_USERNAME,
)
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from .const import (
DEFAULT_OFFSET,
from homeassistant.components import persistent_notification
from homeassistant.components.blink.const import (
DEFAULT_SCAN_INTERVAL,
DEVICE_ID,
DOMAIN,
PLATFORMS,
SERVICE_REFRESH,
SERVICE_SAVE_VIDEO,
SERVICE_SEND_PIN,
)
from homeassistant.const import CONF_FILENAME, CONF_NAME, CONF_PIN, CONF_SCAN_INTERVAL
from homeassistant.core import callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
_LOGGER = logging.getLogger(__name__)
@ -35,58 +28,50 @@ SERVICE_SAVE_VIDEO_SCHEMA = vol.Schema(
)
SERVICE_SEND_PIN_SCHEMA = vol.Schema({vol.Optional(CONF_PIN): cv.string})
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL): int,
}
)
},
extra=vol.ALLOW_EXTRA,
)
def _blink_startup_wrapper(entry):
def _blink_startup_wrapper(hass, entry):
"""Startup wrapper for blink."""
blink = Blink(
username=entry.data[CONF_USERNAME],
password=entry.data[CONF_PASSWORD],
motion_interval=DEFAULT_OFFSET,
legacy_subdomain=False,
no_prompt=True,
device_id=DEVICE_ID,
)
blink = Blink()
auth_data = deepcopy(dict(entry.data))
blink.auth = Auth(auth_data, no_prompt=True)
blink.refresh_rate = entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
try:
blink.login_response = entry.data["login_response"]
blink.setup_params(entry.data["login_response"])
except KeyError:
blink.get_auth_token()
if blink.start():
blink.setup_post_verify()
elif blink.auth.check_key_required():
_LOGGER.debug("Attempting a reauth flow")
_reauth_flow_wrapper(hass, auth_data)
blink.setup_params(entry.data["login_response"])
blink.setup_post_verify()
return blink
async def async_setup(hass, config):
"""Set up a config entry."""
hass.data[DOMAIN] = {}
if DOMAIN not in config:
return True
conf = config.get(DOMAIN, {})
if not hass.config_entries.async_entries(DOMAIN):
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=conf
)
def _reauth_flow_wrapper(hass, data):
"""Reauth flow wrapper."""
hass.add_job(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": "reauth"}, data=data
)
)
persistent_notification.async_create(
hass,
"Blink configuration migrated to a new version. Please go to the integrations page to re-configure (such as sending a new 2FA key).",
"Blink Migration",
)
async def async_setup(hass, config):
"""Set up a Blink component."""
hass.data[DOMAIN] = {}
return True
async def async_migrate_entry(hass, entry):
"""Handle migration of a previous version config entry."""
data = {**entry.data}
if entry.version == 1:
data.pop("login_response", None)
await hass.async_add_executor_job(_reauth_flow_wrapper, hass, data)
return False
return True
@ -95,12 +80,11 @@ async def async_setup_entry(hass, entry):
_async_import_options_from_data_if_missing(hass, entry)
hass.data[DOMAIN][entry.entry_id] = await hass.async_add_executor_job(
_blink_startup_wrapper, entry
_blink_startup_wrapper, hass, entry
)
if not hass.data[DOMAIN][entry.entry_id].available:
_LOGGER.error("Blink unavailable for setup")
return False
raise ConfigEntryNotReady
for component in PLATFORMS:
hass.async_create_task(
@ -118,7 +102,7 @@ async def async_setup_entry(hass, entry):
def send_pin(call):
"""Call blink to send new pin."""
pin = call.data[CONF_PIN]
hass.data[DOMAIN][entry.entry_id].login_handler.send_auth_key(
hass.data[DOMAIN][entry.entry_id].auth.send_auth_key(
hass.data[DOMAIN][entry.entry_id], pin,
)

94
homeassistant/components/blink/config_flow.py
View File

@ -1,10 +1,16 @@
"""Config flow to configure Blink."""
import logging
from blinkpy.blinkpy import Blink
from blinkpy.auth import Auth, LoginError, TokenRefreshFailed
from blinkpy.blinkpy import Blink, BlinkSetupError
import voluptuous as vol
from homeassistant import config_entries, core, exceptions
from homeassistant.components.blink.const import (
DEFAULT_SCAN_INTERVAL,
DEVICE_ID,
DOMAIN,
)
from homeassistant.const import (
CONF_PASSWORD,
CONF_PIN,
@ -13,36 +19,36 @@ from homeassistant.const import (
)
from homeassistant.core import callback
from .const import DEFAULT_OFFSET, DEFAULT_SCAN_INTERVAL, DEVICE_ID, DOMAIN
_LOGGER = logging.getLogger(__name__)
async def validate_input(hass: core.HomeAssistant, blink):
def validate_input(hass: core.HomeAssistant, auth):
"""Validate the user input allows us to connect."""
response = await hass.async_add_executor_job(blink.get_auth_token)
if not response:
try:
auth.startup()
except (LoginError, TokenRefreshFailed):
raise InvalidAuth
if blink.key_required:
if auth.check_key_required():
raise Require2FA
return blink.login_response
def _send_blink_2fa_pin(auth, pin):
"""Send 2FA pin to blink servers."""
blink = Blink()
blink.auth = auth
blink.setup_urls()
return auth.send_auth_key(blink, pin)
class BlinkConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a Blink config flow."""
VERSION = 1
VERSION = 2
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
def __init__(self):
"""Initialize the blink flow."""
self.blink = None
self.data = {
CONF_USERNAME: "",
CONF_PASSWORD: "",
"login_response": None,
}
self.auth = None
@staticmethod
@callback
@ -53,28 +59,19 @@ class BlinkConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_user(self, user_input=None):
"""Handle a flow initiated by the user."""
errors = {}
data = {CONF_USERNAME: "", CONF_PASSWORD: "", "device_id": DEVICE_ID}
if user_input is not None:
self.data[CONF_USERNAME] = user_input["username"]
self.data[CONF_PASSWORD] = user_input["password"]
data[CONF_USERNAME] = user_input["username"]
data[CONF_PASSWORD] = user_input["password"]
await self.async_set_unique_id(self.data[CONF_USERNAME])
if CONF_SCAN_INTERVAL in user_input:
self.data[CONF_SCAN_INTERVAL] = user_input[CONF_SCAN_INTERVAL]
self.blink = Blink(
username=self.data[CONF_USERNAME],
password=self.data[CONF_PASSWORD],
motion_interval=DEFAULT_OFFSET,
legacy_subdomain=False,
no_prompt=True,
device_id=DEVICE_ID,
)
self.auth = Auth(data, no_prompt=True)
await self.async_set_unique_id(data[CONF_USERNAME])
try:
response = await validate_input(self.hass, self.blink)
self.data["login_response"] = response
return self.async_create_entry(title=DOMAIN, data=self.data,)
await self.hass.async_add_executor_job(
validate_input, self.hass, self.auth
)
return self._async_finish_flow()
except Require2FA:
return await self.async_step_2fa()
except InvalidAuth:
@ -94,23 +91,40 @@ class BlinkConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_2fa(self, user_input=None):
"""Handle 2FA step."""
errors = {}
if user_input is not None:
pin = user_input.get(CONF_PIN)
if await self.hass.async_add_executor_job(
self.blink.login_handler.send_auth_key, self.blink, pin
):
return await self.async_step_user(user_input=self.data)
try:
valid_token = await self.hass.async_add_executor_job(
_send_blink_2fa_pin, self.auth, pin
)
except BlinkSetupError:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
if valid_token:
return self._async_finish_flow()
errors["base"] = "invalid_access_token"
return self.async_show_form(
step_id="2fa",
data_schema=vol.Schema(
{vol.Optional("pin"): vol.All(str, vol.Length(min=1))}
),
errors=errors,
)
async def async_step_import(self, import_data):
"""Import blink config from configuration.yaml."""
return await self.async_step_user(import_data)
async def async_step_reauth(self, entry_data):
"""Perform reauth upon migration of old entries."""
return await self.async_step_user(entry_data)
@callback
def _async_finish_flow(self):
"""Finish with setup."""
return self.async_create_entry(title=DOMAIN, data=self.auth.login_attributes)
class BlinkOptionsFlowHandler(config_entries.OptionsFlow):

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

@ -2,6 +2,7 @@
DOMAIN = "blink"
DEVICE_ID = "Home Assistant"
CONF_MIGRATE = "migrate"
CONF_CAMERA = "camera"
CONF_ALARM_CONTROL_PANEL = "alarm_control_panel"

2
homeassistant/components/blink/manifest.json
View File

@ -2,7 +2,7 @@
"domain": "blink",
"name": "Blink",
"documentation": "https://www.home-assistant.io/integrations/blink",
"requirements": ["blinkpy==0.15.1"],
"requirements": ["blinkpy==0.16.3"],
"codeowners": ["@fronzbot"],
"config_flow": true
}

4
homeassistant/components/blink/strings.json
View File

@ -11,11 +11,13 @@
"2fa": {
"title": "Two-factor authentication",
"data": { "2fa": "Two-factor code" },
"description": "Enter the pin sent to your email. If the email does not contain a pin, leave blank"
"description": "Enter the pin sent to your email"
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_access_token": "[%key:common::config_flow::error::invalid_access_token%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {

2
requirements_all.txt
View File

@ -340,7 +340,7 @@ bizkaibus==0.1.1
blebox_uniapi==1.3.2
# homeassistant.components.blink
blinkpy==0.15.1
blinkpy==0.16.3
# homeassistant.components.blinksticklight
blinkstick==1.1.8

2
requirements_test_all.txt
View File

@ -181,7 +181,7 @@ bellows==0.18.0
blebox_uniapi==1.3.2
# homeassistant.components.blink
blinkpy==0.15.1
blinkpy==0.16.3
# homeassistant.components.bom
bomradarloop==0.1.4

238
tests/components/blink/test_config_flow.py
View File

@ -1,4 +1,7 @@
"""Test the Blink config flow."""
from blinkpy.auth import LoginError
from blinkpy.blinkpy import BlinkSetupError
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.blink import DOMAIN
@ -15,13 +18,9 @@ async def test_form(hass):
assert result["type"] == "form"
assert result["errors"] == {}
with patch(
"homeassistant.components.blink.config_flow.Blink",
return_value=Mock(
get_auth_token=Mock(return_value=True),
key_required=False,
login_response={},
),
with patch("homeassistant.components.blink.config_flow.Auth.startup"), patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
), patch(
"homeassistant.components.blink.async_setup", return_value=True
) as mock_setup, patch(
@ -37,48 +36,18 @@ async def test_form(hass):
assert result2["data"] == {
"username": "blink@example.com",
"password": "example",
"login_response": {},
"device_id": "Home Assistant",
"token": None,
"host": None,
"account_id": None,
"client_id": None,
"region_id": None,
}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_import(hass):
"""Test we import the config."""
with patch(
"homeassistant.components.blink.config_flow.Blink",
return_value=Mock(
get_auth_token=Mock(return_value=True),
key_required=False,
login_response={},
),
), patch(
"homeassistant.components.blink.async_setup_entry", return_value=True,
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
"username": "blink@example.com",
"password": "example",
"scan_interval": 10,
},
)
assert result["type"] == "create_entry"
assert result["title"] == "blink"
assert result["result"].unique_id == "blink@example.com"
assert result["data"] == {
"username": "blink@example.com",
"password": "example",
"scan_interval": 10,
"login_response": {},
}
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_2fa(hass):
"""Test we get the 2fa form."""
await setup.async_setup_component(hass, "persistent_notification", {})
@ -86,28 +55,28 @@ async def test_form_2fa(hass):
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_blink = Mock(
get_auth_token=Mock(return_value=True),
key_required=True,
login_response={},
login_handler=Mock(send_auth_key=Mock(return_value=True)),
)
with patch(
"homeassistant.components.blink.config_flow.Blink", return_value=mock_blink
with patch("homeassistant.components.blink.config_flow.Auth.startup"), patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=True,
), patch(
"homeassistant.components.blink.async_setup", return_value=True
) as mock_setup:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"username": "blink@example.com", "password": "example"}
result["flow_id"], {"username": "blink@example.com", "password": "example"},
)
assert result2["type"] == "form"
assert result2["step_id"] == "2fa"
mock_blink.key_required = False
with patch(
"homeassistant.components.blink.config_flow.Blink", return_value=mock_blink
with patch("homeassistant.components.blink.config_flow.Auth.startup"), patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
), patch(
"homeassistant.components.blink.config_flow.Auth.send_auth_key",
return_value=True,
), patch(
"homeassistant.components.blink.config_flow.Blink.setup_urls",
return_value=True,
), patch(
"homeassistant.components.blink.async_setup", return_value=True
) as mock_setup, patch(
@ -125,6 +94,126 @@ async def test_form_2fa(hass):
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_2fa_connect_error(hass):
"""Test we report a connect error during 2fa setup."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch("homeassistant.components.blink.config_flow.Auth.startup"), patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=True,
), patch("homeassistant.components.blink.async_setup", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"username": "blink@example.com", "password": "example"},
)
assert result2["type"] == "form"
assert result2["step_id"] == "2fa"
with patch("homeassistant.components.blink.config_flow.Auth.startup"), patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
), patch(
"homeassistant.components.blink.config_flow.Auth.send_auth_key",
return_value=True,
), patch(
"homeassistant.components.blink.config_flow.Blink.setup_urls",
side_effect=BlinkSetupError,
), patch(
"homeassistant.components.blink.async_setup", return_value=True
), patch(
"homeassistant.components.blink.async_setup_entry", return_value=True
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"], {"pin": "1234"}
)
assert result3["type"] == "form"
assert result3["errors"] == {"base": "cannot_connect"}
async def test_form_2fa_invalid_key(hass):
"""Test we report an error if key is invalid."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch("homeassistant.components.blink.config_flow.Auth.startup"), patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=True,
), patch("homeassistant.components.blink.async_setup", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"username": "blink@example.com", "password": "example"},
)
assert result2["type"] == "form"
assert result2["step_id"] == "2fa"
with patch("homeassistant.components.blink.config_flow.Auth.startup",), patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
), patch(
"homeassistant.components.blink.config_flow.Auth.send_auth_key",
return_value=False,
), patch(
"homeassistant.components.blink.config_flow.Blink.setup_urls",
return_value=True,
), patch(
"homeassistant.components.blink.async_setup", return_value=True
), patch(
"homeassistant.components.blink.async_setup_entry", return_value=True
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"], {"pin": "1234"}
)
assert result3["type"] == "form"
assert result3["errors"] == {"base": "invalid_access_token"}
async def test_form_2fa_unknown_error(hass):
"""Test we report an unknown error during 2fa setup."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch("homeassistant.components.blink.config_flow.Auth.startup"), patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=True,
), patch("homeassistant.components.blink.async_setup", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"username": "blink@example.com", "password": "example"},
)
assert result2["type"] == "form"
assert result2["step_id"] == "2fa"
with patch("homeassistant.components.blink.config_flow.Auth.startup"), patch(
"homeassistant.components.blink.config_flow.Auth.check_key_required",
return_value=False,
), patch(
"homeassistant.components.blink.config_flow.Auth.send_auth_key",
return_value=True,
), patch(
"homeassistant.components.blink.config_flow.Blink.setup_urls",
side_effect=KeyError,
), patch(
"homeassistant.components.blink.async_setup", return_value=True
), patch(
"homeassistant.components.blink.async_setup_entry", return_value=True
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"], {"pin": "1234"}
)
assert result3["type"] == "form"
assert result3["errors"] == {"base": "unknown"}
async def test_form_invalid_auth(hass):
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
@ -132,8 +221,8 @@ async def test_form_invalid_auth(hass):
)
with patch(
"homeassistant.components.blink.config_flow.Blink.get_auth_token",
return_value=None,
"homeassistant.components.blink.config_flow.Auth.startup",
side_effect=LoginError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"username": "blink@example.com", "password": "example"}
@ -150,11 +239,7 @@ async def test_form_unknown_error(hass):
)
with patch(
"homeassistant.components.blink.config_flow.Blink.get_auth_token",
return_value=None,
), patch(
"homeassistant.components.blink.config_flow.validate_input",
side_effect=KeyError,
"homeassistant.components.blink.config_flow.Auth.startup", side_effect=KeyError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {"username": "blink@example.com", "password": "example"}
@ -164,27 +249,34 @@ async def test_form_unknown_error(hass):
assert result2["errors"] == {"base": "unknown"}
async def test_reauth_shows_user_step(hass):
"""Test reauth shows the user form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "reauth"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
async def test_options_flow(hass):
"""Test config flow options."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={
"username": "blink@example.com",
"password": "example",
"login_response": {},
},
data={"username": "blink@example.com", "password": "example"},
options={},
entry_id=1,
version=2,
)
config_entry.add_to_hass(hass)
mock_blink = Mock(
login_handler=True,
setup_params=Mock(return_value=True),
setup_post_verify=Mock(return_value=True),
mock_auth = Mock(
startup=Mock(return_value=True), check_key_required=Mock(return_value=False)
)
mock_blink = Mock()
with patch("homeassistant.components.blink.Blink", return_value=mock_blink):
with patch("homeassistant.components.blink.Auth", return_value=mock_auth), patch(
"homeassistant.components.blink.Blink", return_value=mock_blink
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()