From 53738c0168b99a534cd87ed4611976b195cd98b9 Mon Sep 17 00:00:00 2001 From: Fabian Neundorf Date: Thu, 9 Apr 2026 23:09:35 +0200 Subject: [PATCH] Add 2fa support in picnic integration (#167636) --- .../components/picnic/config_flow.py | 210 +++++++---- homeassistant/components/picnic/const.py | 1 + homeassistant/components/picnic/strings.json | 23 ++ tests/components/picnic/test_config_flow.py | 339 +++++++++++++++--- 4 files changed, 456 insertions(+), 117 deletions(-) diff --git a/homeassistant/components/picnic/config_flow.py b/homeassistant/components/picnic/config_flow.py index a60086173a84..118d752dd9ec 100644 --- a/homeassistant/components/picnic/config_flow.py +++ b/homeassistant/components/picnic/config_flow.py @@ -7,7 +7,11 @@ import logging from typing import Any from python_picnic_api2 import PicnicAPI -from python_picnic_api2.session import PicnicAuthError +from python_picnic_api2.session import ( + Picnic2FAError, + Picnic2FARequired, + PicnicAuthError, +) import requests import voluptuous as vol @@ -18,13 +22,19 @@ from homeassistant.const import ( CONF_PASSWORD, CONF_USERNAME, ) -from homeassistant.core import HomeAssistant -from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.selector import ( + SelectSelector, + SelectSelectorConfig, + SelectSelectorMode, +) -from .const import COUNTRY_CODES, DOMAIN +from .const import COUNTRY_CODES, DOMAIN, TWO_FA_CHANNELS _LOGGER = logging.getLogger(__name__) +CONF_2FA_CODE = "two_fa_code" +CONF_2FA_CHANNEL = "two_fa_channel" + STEP_USER_DATA_SCHEMA = vol.Schema( { vol.Required(CONF_USERNAME): str, @@ -35,45 +45,23 @@ STEP_USER_DATA_SCHEMA = vol.Schema( } ) - -class PicnicHub: - """Hub class to test user authentication.""" - - @staticmethod - def authenticate(username, password, country_code) -> tuple[str, dict]: - """Test if we can authenticate with the Picnic API.""" - picnic = PicnicAPI(username, password, country_code) - return picnic.session.auth_token, picnic.get_user() - - -async def validate_input(hass: HomeAssistant, data): - """Validate the user input allows us to connect. - - Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user. - """ - hub = PicnicHub() - - try: - auth_token, user_data = await hass.async_add_executor_job( - hub.authenticate, - data[CONF_USERNAME], - data[CONF_PASSWORD], - data[CONF_COUNTRY_CODE], - ) - except requests.exceptions.ConnectionError as error: - raise CannotConnect from error - except PicnicAuthError as error: - raise InvalidAuth from error - - # Return the validation result - address = ( - f"{user_data['address']['street']} {user_data['address']['house_number']}" - f"{user_data['address']['house_number_ext']}" - ) - return auth_token, { - "title": address, - "unique_id": user_data["user_id"], +STEP_2FA_CHANNEL_SCHEMA = vol.Schema( + { + vol.Required(CONF_2FA_CHANNEL, default=TWO_FA_CHANNELS[0]): SelectSelector( + SelectSelectorConfig( + options=TWO_FA_CHANNELS, + mode=SelectSelectorMode.LIST, + translation_key="two_fa_channel", + ) + ), } +) + +STEP_2FA_SCHEMA = vol.Schema( + { + vol.Required(CONF_2FA_CODE): str, + } +) class PicnicConfigFlow(ConfigFlow, domain=DOMAIN): @@ -81,6 +69,11 @@ class PicnicConfigFlow(ConfigFlow, domain=DOMAIN): VERSION = 1 + def __init__(self) -> None: + """Initialize the config flow.""" + self._picnic: PicnicAPI | None = None + self._user_input: dict[str, Any] = {} + async def async_step_reauth( self, entry_data: Mapping[str, Any] ) -> ConfigFlowResult: @@ -90,7 +83,7 @@ class PicnicConfigFlow(ConfigFlow, domain=DOMAIN): async def async_step_user( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: - """Handle the authentication step, this is the generic step for both `step_user` and `step_reauth`.""" + """Handle the authentication step.""" if user_input is None: return self.async_show_form( step_id="user", data_schema=STEP_USER_DATA_SCHEMA @@ -99,43 +92,122 @@ class PicnicConfigFlow(ConfigFlow, domain=DOMAIN): errors = {} try: - auth_token, info = await validate_input(self.hass, user_input) - except CannotConnect: + await self.hass.async_add_executor_job( + self._start_login, + user_input[CONF_USERNAME], + user_input[CONF_PASSWORD], + user_input[CONF_COUNTRY_CODE], + ) + except Picnic2FARequired: + self._user_input = user_input + return await self.async_step_2fa_channel() + except requests.exceptions.ConnectionError: errors["base"] = "cannot_connect" - except InvalidAuth: + except PicnicAuthError: errors["base"] = "invalid_auth" except Exception: _LOGGER.exception("Unexpected exception") errors["base"] = "unknown" else: - data = { - CONF_ACCESS_TOKEN: auth_token, - CONF_COUNTRY_CODE: user_input[CONF_COUNTRY_CODE], - } - existing_entry = await self.async_set_unique_id(info["unique_id"]) - - # Abort if we're adding a new config and the unique id is already in use, else create the entry - if self.source != SOURCE_REAUTH: - self._abort_if_unique_id_configured() - return self.async_create_entry(title="Picnic", data=data) - - # In case of re-auth, only continue if an exiting account exists with the same unique id - if existing_entry: - self.hass.config_entries.async_update_entry(existing_entry, data=data) - await self.hass.config_entries.async_reload(existing_entry.entry_id) - return self.async_abort(reason="reauth_successful") - - # Set the error because the account is different - errors["base"] = "different_account" + return await self._async_finish(user_input) return self.async_show_form( step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors ) + def _start_login(self, username: str, password: str, country_code: str) -> None: + self._picnic = PicnicAPI(country_code=country_code) + self._picnic.login(username, password) -class CannotConnect(HomeAssistantError): - """Error to indicate we cannot connect.""" + async def async_step_2fa_channel( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Let the user pick the 2FA delivery channel.""" + assert self._picnic is not None + if user_input is None: + return self.async_show_form( + step_id="2fa_channel", data_schema=STEP_2FA_CHANNEL_SCHEMA + ) -class InvalidAuth(HomeAssistantError): - """Error to indicate there is invalid auth.""" + errors = {} + channel = user_input[CONF_2FA_CHANNEL].upper() + try: + await self.hass.async_add_executor_job( + self._picnic.generate_2fa_code, channel + ) + except requests.exceptions.ConnectionError: + errors["base"] = "cannot_connect" + except Exception: + _LOGGER.exception("Failed to request 2FA code via %s", channel) + errors["base"] = "unknown" + else: + return await self.async_step_2fa() + + return self.async_show_form( + step_id="2fa_channel", + data_schema=STEP_2FA_CHANNEL_SCHEMA, + errors=errors, + ) + + async def async_step_2fa( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle the 2FA verification step.""" + assert self._picnic is not None + + if user_input is None: + return self.async_show_form(step_id="2fa", data_schema=STEP_2FA_SCHEMA) + + errors = {} + + try: + await self.hass.async_add_executor_job( + self._picnic.verify_2fa_code, user_input[CONF_2FA_CODE] + ) + except Picnic2FAError: + errors["base"] = "invalid_2fa_code" + except requests.exceptions.ConnectionError: + errors["base"] = "cannot_connect" + except Exception: + _LOGGER.exception("Unexpected exception during 2FA verification") + errors["base"] = "unknown" + else: + return await self._async_finish(self._user_input) + + return self.async_show_form( + step_id="2fa", data_schema=STEP_2FA_SCHEMA, errors=errors + ) + + async def _async_finish( + self, + user_input: dict[str, Any], + ) -> ConfigFlowResult: + """Finalize the config entry after successful authentication.""" + assert self._picnic is not None + + auth_token = self._picnic.session.auth_token + user_data = await self.hass.async_add_executor_job(self._picnic.get_user) + + data = { + CONF_ACCESS_TOKEN: auth_token, + CONF_COUNTRY_CODE: user_input[CONF_COUNTRY_CODE], + } + existing_entry = await self.async_set_unique_id(user_data["user_id"]) + + # Abort if we're adding a new config and the unique id is already in use, else create the entry + if self.source != SOURCE_REAUTH: + self._abort_if_unique_id_configured() + return self.async_create_entry(title="Picnic", data=data) + + # In case of re-auth, only continue if an exiting account exists with the same unique id + if existing_entry: + self.hass.config_entries.async_update_entry(existing_entry, data=data) + await self.hass.config_entries.async_reload(existing_entry.entry_id) + return self.async_abort(reason="reauth_successful") + + return self.async_show_form( + step_id="user", + data_schema=STEP_USER_DATA_SCHEMA, + errors={"base": "different_account"}, + ) diff --git a/homeassistant/components/picnic/const.py b/homeassistant/components/picnic/const.py index 9cde3dea03dc..8996e5b971b6 100644 --- a/homeassistant/components/picnic/const.py +++ b/homeassistant/components/picnic/const.py @@ -12,6 +12,7 @@ ATTR_AMOUNT = "amount" ATTR_PRODUCT_IDENTIFIERS = "product_identifiers" COUNTRY_CODES = ["NL", "DE", "BE", "FR"] +TWO_FA_CHANNELS = ["sms", "email"] ATTRIBUTION = "Data provided by Picnic" ADDRESS = "address" CART_DATA = "cart_data" diff --git a/homeassistant/components/picnic/strings.json b/homeassistant/components/picnic/strings.json index db56d032b1d2..e2cea9b4d4d2 100644 --- a/homeassistant/components/picnic/strings.json +++ b/homeassistant/components/picnic/strings.json @@ -7,10 +7,25 @@ "error": { "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", "different_account": "Account should be the same as used for setting up the integration", + "invalid_2fa_code": "The verification code is incorrect or has expired.", "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]", "unknown": "[%key:common::config_flow::error::unknown%]" }, "step": { + "2fa": { + "data": { + "two_fa_code": "Verification code" + }, + "description": "A verification code has been sent to you via your selected channel.", + "title": "Two-factor authentication" + }, + "2fa_channel": { + "data": { + "two_fa_channel": "Channel" + }, + "description": "A second factor is required to complete the login. Select the channel through which you want to receive your second factor.", + "title": "Two-factor authentication" + }, "user": { "data": { "country_code": "Country code", @@ -77,6 +92,14 @@ } } }, + "selector": { + "two_fa_channel": { + "options": { + "email": "Email", + "sms": "Text message (SMS)" + } + } + }, "services": { "add_product": { "description": "Adds a product to the cart based on a search string or product ID. The search string and product ID are exclusive.", diff --git a/tests/components/picnic/test_config_flow.py b/tests/components/picnic/test_config_flow.py index ba4c36682e1f..63c03820f60c 100644 --- a/tests/components/picnic/test_config_flow.py +++ b/tests/components/picnic/test_config_flow.py @@ -3,7 +3,11 @@ from unittest.mock import patch import pytest -from python_picnic_api2.session import PicnicAuthError +from python_picnic_api2.session import ( + Picnic2FAError, + Picnic2FARequired, + PicnicAuthError, +) import requests from homeassistant import config_entries @@ -30,8 +34,12 @@ def picnic_api(): with patch( "homeassistant.components.picnic.config_flow.PicnicAPI", ) as picnic_mock: - picnic_mock().session.auth_token = auth_token - picnic_mock().get_user.return_value = auth_data + instance = picnic_mock.return_value + instance.session.auth_token = auth_token + instance.get_user.return_value = auth_data + instance.login.return_value = None # no 2FA by default + instance.generate_2fa_code.return_value = None + instance.verify_2fa_code.return_value = None yield picnic_mock @@ -69,17 +77,19 @@ async def test_form(hass: HomeAssistant, picnic_api) -> None: assert len(mock_setup_entry.mock_calls) == 1 -async def test_form_invalid_auth(hass: HomeAssistant) -> None: - """Test we handle invalid authentication.""" +async def test_form_2fa_required(hass: HomeAssistant, picnic_api) -> None: + """Test the full 2FA flow.""" + picnic_api.return_value.login.side_effect = Picnic2FARequired + result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER} ) with patch( - "homeassistant.components.picnic.config_flow.PicnicHub.authenticate", - side_effect=PicnicAuthError, + "homeassistant.components.picnic.async_setup_entry", + return_value=True, ): - result2 = await hass.config_entries.flow.async_configure( + result_step_user = await hass.config_entries.flow.async_configure( result["flow_id"], { "username": "test-username", @@ -87,52 +97,287 @@ async def test_form_invalid_auth(hass: HomeAssistant) -> None: "country_code": "NL", }, ) + assert result_step_user["type"] is FlowResultType.FORM + assert result_step_user["step_id"] == "2fa_channel" + + result_step_2fa_channel = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_channel": "sms"}, + ) + assert result_step_2fa_channel["type"] is FlowResultType.FORM + assert result_step_2fa_channel["step_id"] == "2fa" + + result_step_2fa_verify = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_code": "123456"}, + ) + await hass.async_block_till_done() + + assert result_step_2fa_verify["type"] is FlowResultType.CREATE_ENTRY + assert result_step_2fa_verify["title"] == "Picnic" + assert result_step_2fa_verify["data"] == { + CONF_ACCESS_TOKEN: picnic_api().session.auth_token, + CONF_COUNTRY_CODE: "NL", + } + assert picnic_api.return_value.generate_2fa_code.call_count == 1 + assert picnic_api.return_value.generate_2fa_code.call_args[0] == ("SMS",) + assert picnic_api.return_value.verify_2fa_code.call_count == 1 + assert picnic_api.return_value.verify_2fa_code.call_args[0] == ("123456",) + + +async def test_form_2fa_channel_cannot_connect(hass: HomeAssistant, picnic_api) -> None: + """Test we handle connection errors in the first 2fa step.""" + picnic_api.return_value.login.side_effect = Picnic2FARequired + picnic_api.return_value.generate_2fa_code.side_effect = ( + requests.exceptions.ConnectionError + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.picnic.async_setup_entry", + return_value=True, + ): + result_step_user = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "username": "test-username", + "password": "test-password", + "country_code": "NL", + }, + ) + assert result_step_user["type"] is FlowResultType.FORM + assert result_step_user["step_id"] == "2fa_channel" + + result_step_2fa_channel = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_channel": "sms"}, + ) + await hass.async_block_till_done() + + assert result_step_2fa_channel["type"] is FlowResultType.FORM + assert result_step_2fa_channel["errors"] == {"base": "cannot_connect"} + + +async def test_form_2fa_channel_exception(hass: HomeAssistant, picnic_api) -> None: + """Test we handle random exceptions in the first 2fa step.""" + picnic_api.return_value.login.side_effect = Picnic2FARequired + picnic_api.return_value.generate_2fa_code.side_effect = Exception + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.picnic.async_setup_entry", + return_value=True, + ): + result_step_user = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "username": "test-username", + "password": "test-password", + "country_code": "NL", + }, + ) + assert result_step_user["type"] is FlowResultType.FORM + assert result_step_user["step_id"] == "2fa_channel" + + result_step_2fa_channel = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_channel": "sms"}, + ) + await hass.async_block_till_done() + + assert result_step_2fa_channel["type"] is FlowResultType.FORM + assert result_step_2fa_channel["errors"] == {"base": "unknown"} + + +async def test_form_2fa_wrong_code(hass: HomeAssistant, picnic_api) -> None: + """Test the full 2FA flow with incorrect code.""" + picnic_api.return_value.login.side_effect = Picnic2FARequired + picnic_api.return_value.verify_2fa_code.side_effect = Picnic2FAError + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.picnic.async_setup_entry", + return_value=True, + ): + result_step_user = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "username": "test-username", + "password": "test-password", + "country_code": "NL", + }, + ) + assert result_step_user["type"] is FlowResultType.FORM + assert result_step_user["step_id"] == "2fa_channel" + + result_step_2fa_channel = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_channel": "sms"}, + ) + assert result_step_2fa_channel["type"] is FlowResultType.FORM + assert result_step_2fa_channel["step_id"] == "2fa" + + result_step_2fa_verify = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_code": "654321"}, + ) + await hass.async_block_till_done() + + assert result_step_2fa_verify["type"] is FlowResultType.FORM + assert result_step_2fa_verify["errors"] == {"base": "invalid_2fa_code"} + + +async def test_form_2fa_cannot_connect(hass: HomeAssistant, picnic_api) -> None: + """Test we handle connection errors in the last 2fa step.""" + picnic_api.return_value.login.side_effect = Picnic2FARequired + picnic_api.return_value.verify_2fa_code.side_effect = ( + requests.exceptions.ConnectionError + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.picnic.async_setup_entry", + return_value=True, + ): + result_step_user = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "username": "test-username", + "password": "test-password", + "country_code": "NL", + }, + ) + assert result_step_user["type"] is FlowResultType.FORM + assert result_step_user["step_id"] == "2fa_channel" + + result_step_2fa_channel = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_channel": "sms"}, + ) + assert result_step_2fa_channel["type"] is FlowResultType.FORM + assert result_step_2fa_channel["step_id"] == "2fa" + + result_step_2fa_verify = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_code": "123456"}, + ) + await hass.async_block_till_done() + + assert result_step_2fa_verify["type"] is FlowResultType.FORM + assert result_step_2fa_verify["errors"] == {"base": "cannot_connect"} + + +async def test_form_2fa_exception(hass: HomeAssistant, picnic_api) -> None: + """Test we handle random exceptions in the last 2fa step.""" + picnic_api.return_value.login.side_effect = Picnic2FARequired + picnic_api.return_value.verify_2fa_code.side_effect = Exception + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.picnic.async_setup_entry", + return_value=True, + ): + result_step_user = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "username": "test-username", + "password": "test-password", + "country_code": "NL", + }, + ) + assert result_step_user["type"] is FlowResultType.FORM + assert result_step_user["step_id"] == "2fa_channel" + + result_step_2fa_channel = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_channel": "sms"}, + ) + assert result_step_2fa_channel["type"] is FlowResultType.FORM + assert result_step_2fa_channel["step_id"] == "2fa" + + result_step_2fa_verify = await hass.config_entries.flow.async_configure( + result["flow_id"], + {"two_fa_code": "123456"}, + ) + await hass.async_block_till_done() + + assert result_step_2fa_verify["type"] is FlowResultType.FORM + assert result_step_2fa_verify["errors"] == {"base": "unknown"} + + +async def test_form_invalid_auth(hass: HomeAssistant, picnic_api) -> None: + """Test we handle invalid authentication.""" + picnic_api.return_value.login.side_effect = PicnicAuthError + + 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"], + { + "username": "test-username", + "password": "test-password", + "country_code": "NL", + }, + ) assert result2["type"] is FlowResultType.FORM assert result2["errors"] == {"base": "invalid_auth"} -async def test_form_cannot_connect(hass: HomeAssistant) -> None: +async def test_form_cannot_connect(hass: HomeAssistant, picnic_api) -> None: """Test we handle connection errors.""" + picnic_api.return_value.login.side_effect = requests.exceptions.ConnectionError + result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER} ) - with patch( - "homeassistant.components.picnic.config_flow.PicnicHub.authenticate", - side_effect=requests.exceptions.ConnectionError, - ): - result2 = await hass.config_entries.flow.async_configure( - result["flow_id"], - { - "username": "test-username", - "password": "test-password", - "country_code": "NL", - }, - ) + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "username": "test-username", + "password": "test-password", + "country_code": "NL", + }, + ) assert result2["type"] is FlowResultType.FORM assert result2["errors"] == {"base": "cannot_connect"} -async def test_form_exception(hass: HomeAssistant) -> None: +async def test_form_exception(hass: HomeAssistant, picnic_api) -> None: """Test we handle random exceptions.""" + picnic_api.return_value.login.side_effect = Exception + result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER} ) - with patch( - "homeassistant.components.picnic.config_flow.PicnicHub.authenticate", - side_effect=Exception, - ): - result2 = await hass.config_entries.flow.async_configure( - result["flow_id"], - { - "username": "test-username", - "password": "test-password", - "country_code": "NL", - }, - ) + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "username": "test-username", + "password": "test-password", + "country_code": "NL", + }, + ) assert result2["type"] is FlowResultType.FORM assert result2["errors"] == {"base": "unknown"} @@ -203,8 +448,10 @@ async def test_step_reauth(hass: HomeAssistant, picnic_api) -> None: assert len(hass.config_entries.async_entries()) == 1 -async def test_step_reauth_failed(hass: HomeAssistant) -> None: +async def test_step_reauth_failed(hass: HomeAssistant, picnic_api) -> None: """Test the re-auth flow when authentication fails.""" + picnic_api.return_value.login.side_effect = PicnicAuthError + # Create a mocked config entry user_id = "f29-2a6-o32n" conf = {CONF_ACCESS_TOKEN: "a3p98fsen.a39p3fap", CONF_COUNTRY_CODE: "NL"} @@ -221,19 +468,15 @@ async def test_step_reauth_failed(hass: HomeAssistant) -> None: assert result_init["type"] is FlowResultType.FORM assert result_init["step_id"] == "user" - with patch( - "homeassistant.components.picnic.config_flow.PicnicHub.authenticate", - side_effect=PicnicAuthError, - ): - result_configure = await hass.config_entries.flow.async_configure( - result_init["flow_id"], - { - "username": "test-username", - "password": "test-password", - "country_code": "NL", - }, - ) - await hass.async_block_till_done() + result_configure = await hass.config_entries.flow.async_configure( + result_init["flow_id"], + { + "username": "test-username", + "password": "test-password", + "country_code": "NL", + }, + ) + await hass.async_block_till_done() # Check that the returned flow has type form with error set assert result_configure["type"] is FlowResultType.FORM