diff --git a/homeassistant/components/huawei_lte/__init__.py b/homeassistant/components/huawei_lte/__init__.py
index d0d1ce71161..62efabf1f5e 100644
--- a/homeassistant/components/huawei_lte/__init__.py
+++ b/homeassistant/components/huawei_lte/__init__.py
@@ -35,6 +35,7 @@ from homeassistant.const import (
CONF_RECIPIENT,
CONF_URL,
CONF_USERNAME,
+ CONF_VERIFY_SSL,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
@@ -89,7 +90,7 @@ from .const import (
SERVICE_SUSPEND_INTEGRATION,
UPDATE_SIGNAL,
)
-from .utils import get_device_macs
+from .utils import get_device_macs, non_verifying_requests_session
_LOGGER = logging.getLogger(__name__)
@@ -335,16 +336,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
def _connect() -> Connection:
"""Set up a connection."""
+ kwargs: dict[str, Any] = {
+ "timeout": CONNECTION_TIMEOUT,
+ }
+ if url.startswith("https://") and not entry.data.get(CONF_VERIFY_SSL):
+ kwargs["requests_session"] = non_verifying_requests_session(url)
if entry.options.get(CONF_UNAUTHENTICATED_MODE):
_LOGGER.debug("Connecting in unauthenticated mode, reduced feature set")
- connection = Connection(url, timeout=CONNECTION_TIMEOUT)
+ connection = Connection(url, **kwargs)
else:
_LOGGER.debug("Connecting in authenticated mode, full feature set")
username = entry.data.get(CONF_USERNAME) or ""
password = entry.data.get(CONF_PASSWORD) or ""
- connection = Connection(
- url, username=username, password=password, timeout=CONNECTION_TIMEOUT
- )
+ connection = Connection(url, username=username, password=password, **kwargs)
return connection
try:
diff --git a/homeassistant/components/huawei_lte/config_flow.py b/homeassistant/components/huawei_lte/config_flow.py
index 6d7b0b9bb11..c97c8d6367b 100644
--- a/homeassistant/components/huawei_lte/config_flow.py
+++ b/homeassistant/components/huawei_lte/config_flow.py
@@ -16,7 +16,7 @@ from huawei_lte_api.exceptions import (
ResponseErrorException,
)
from huawei_lte_api.Session import GetResponseType
-from requests.exceptions import Timeout
+from requests.exceptions import SSLError, Timeout
from url_normalize import url_normalize
import voluptuous as vol
@@ -29,6 +29,7 @@ from homeassistant.const import (
CONF_RECIPIENT,
CONF_URL,
CONF_USERNAME,
+ CONF_VERIFY_SSL,
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
@@ -44,7 +45,7 @@ from .const import (
DEFAULT_UNAUTHENTICATED_MODE,
DOMAIN,
)
-from .utils import get_device_macs
+from .utils import get_device_macs, non_verifying_requests_session
_LOGGER = logging.getLogger(__name__)
@@ -80,6 +81,13 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self.context.get(CONF_URL, ""),
),
): str,
+ vol.Optional(
+ CONF_VERIFY_SSL,
+ default=user_input.get(
+ CONF_VERIFY_SSL,
+ False,
+ ),
+ ): bool,
vol.Optional(
CONF_USERNAME, default=user_input.get(CONF_USERNAME) or ""
): str,
@@ -119,11 +127,20 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
password = user_input.get(CONF_PASSWORD) or ""
def _get_connection() -> Connection:
+ if (
+ user_input[CONF_URL].startswith("https://")
+ and not user_input[CONF_VERIFY_SSL]
+ ):
+ requests_session = non_verifying_requests_session(user_input[CONF_URL])
+ else:
+ requests_session = None
+
return Connection(
url=user_input[CONF_URL],
username=username,
password=password,
timeout=CONNECTION_TIMEOUT,
+ requests_session=requests_session,
)
conn = None
@@ -140,6 +157,12 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
except ResponseErrorException:
_LOGGER.warning("Response error", exc_info=True)
errors["base"] = "response_error"
+ except SSLError:
+ _LOGGER.warning("SSL error", exc_info=True)
+ if user_input[CONF_VERIFY_SSL]:
+ errors[CONF_URL] = "ssl_error_try_unverified"
+ else:
+ errors[CONF_URL] = "ssl_error_try_plain"
except Timeout:
_LOGGER.warning("Connection timeout", exc_info=True)
errors[CONF_URL] = "connection_timeout"
@@ -152,6 +175,7 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def _disconnect(conn: Connection) -> None:
try:
conn.close()
+ conn.requests_session.close()
except Exception: # pylint: disable=broad-except
_LOGGER.debug("Disconnect error", exc_info=True)
diff --git a/homeassistant/components/huawei_lte/strings.json b/homeassistant/components/huawei_lte/strings.json
index 1e43aa818e9..9e46ca742b8 100644
--- a/homeassistant/components/huawei_lte/strings.json
+++ b/homeassistant/components/huawei_lte/strings.json
@@ -14,6 +14,8 @@
"invalid_url": "Invalid URL",
"login_attempts_exceeded": "Maximum login attempts exceeded, please try again later",
"response_error": "Unknown error from device",
+ "ssl_error_try_plain": "HTTPS error, please try a plain HTTP URL",
+ "ssl_error_try_unverified": "HTTPS error, please try disabling certificate verification or a plain HTTP URL",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"flow_title": "{name}",
@@ -30,7 +32,8 @@
"data": {
"password": "[%key:common::config_flow::data::password%]",
"url": "[%key:common::config_flow::data::url%]",
- "username": "[%key:common::config_flow::data::username%]"
+ "username": "[%key:common::config_flow::data::username%]",
+ "verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
},
"description": "Enter device access details.",
"title": "Configure Huawei LTE"
diff --git a/homeassistant/components/huawei_lte/utils.py b/homeassistant/components/huawei_lte/utils.py
index 172e8658928..df212a1c25d 100644
--- a/homeassistant/components/huawei_lte/utils.py
+++ b/homeassistant/components/huawei_lte/utils.py
@@ -2,8 +2,13 @@
from __future__ import annotations
from contextlib import suppress
+import re
+from urllib.parse import urlparse
+import warnings
from huawei_lte_api.Session import GetResponseType
+import requests
+from urllib3.exceptions import InsecureRequestWarning
from homeassistant.helpers.device_registry import format_mac
@@ -25,3 +30,18 @@ def get_device_macs(
macs.extend(x.get("WifiMac") for x in wlan_settings["Ssids"]["Ssid"])
return sorted({format_mac(str(x)) for x in macs if x})
+
+
+def non_verifying_requests_session(url: str) -> requests.Session:
+ """Get requests.Session that does not verify HTTPS, filter warnings about it."""
+ parsed_url = urlparse(url)
+ assert parsed_url.hostname
+ requests_session = requests.Session()
+ requests_session.verify = False
+ warnings.filterwarnings(
+ "ignore",
+ message=rf"^.*\b{re.escape(parsed_url.hostname)}\b",
+ category=InsecureRequestWarning,
+ module=r"^urllib3\.connectionpool$",
+ )
+ return requests_session
diff --git a/tests/components/huawei_lte/test_config_flow.py b/tests/components/huawei_lte/test_config_flow.py
index 13307e43648..e358920b07b 100644
--- a/tests/components/huawei_lte/test_config_flow.py
+++ b/tests/components/huawei_lte/test_config_flow.py
@@ -1,5 +1,7 @@
"""Tests for the Huawei LTE config flow."""
+from typing import Any
from unittest.mock import patch
+from urllib.parse import urlparse, urlunparse
from huawei_lte_api.enums.client import ResponseCodeEnum
from huawei_lte_api.enums.user import LoginErrorEnum, LoginStateEnum, PasswordTypeEnum
@@ -18,6 +20,7 @@ from homeassistant.const import (
CONF_RECIPIENT,
CONF_URL,
CONF_USERNAME,
+ CONF_VERIFY_SSL,
)
from homeassistant.core import HomeAssistant
@@ -25,8 +28,9 @@ from tests.common import MockConfigEntry
FIXTURE_UNIQUE_ID = "SERIALNUMBER"
-FIXTURE_USER_INPUT = {
+FIXTURE_USER_INPUT: dict[str, Any] = {
CONF_URL: "http://192.168.1.1/",
+ CONF_VERIFY_SSL: False,
CONF_USERNAME: "admin",
CONF_PASSWORD: "secret",
}
@@ -95,34 +99,59 @@ async def test_already_configured(
assert result["reason"] == "already_configured"
-async def test_connection_error(
- hass: HomeAssistant, requests_mock: requests_mock.Mocker
-) -> None:
- """Test we show user form on connection error."""
- requests_mock.request(ANY, ANY, exc=ConnectionError())
+@pytest.mark.parametrize(
+ ("exception", "errors", "data_patch"),
+ (
+ (ConnectionError(), {CONF_URL: "unknown"}, {}),
+ (requests.exceptions.SSLError(), {CONF_URL: "ssl_error_try_plain"}, {}),
+ (
+ requests.exceptions.SSLError(),
+ {CONF_URL: "ssl_error_try_unverified"},
+ {CONF_VERIFY_SSL: True},
+ ),
+ ),
+)
+async def test_connection_errors(
+ hass: HomeAssistant,
+ requests_mock: requests_mock.Mocker,
+ exception: Exception,
+ errors: dict[str, str],
+ data_patch: dict[str, Any],
+):
+ """Test we show user form on various errors."""
+ requests_mock.request(ANY, ANY, exc=exception)
result = await hass.config_entries.flow.async_init(
- DOMAIN, context={"source": config_entries.SOURCE_USER}, data=FIXTURE_USER_INPUT
+ DOMAIN,
+ context={"source": config_entries.SOURCE_USER},
+ data=FIXTURE_USER_INPUT | data_patch,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
- assert result["errors"] == {CONF_URL: "unknown"}
+ assert result["errors"] == errors
@pytest.fixture
def login_requests_mock(requests_mock):
"""Set up a requests_mock with base mocks for login tests."""
- requests_mock.request(
- ANY, FIXTURE_USER_INPUT[CONF_URL], text=''
- )
- requests_mock.request(
- ANY,
- f"{FIXTURE_USER_INPUT[CONF_URL]}api/user/state-login",
- text=(
- f"{LoginStateEnum.LOGGED_OUT}"
- f"{PasswordTypeEnum.SHA256}"
- ),
+ https_url = urlunparse(
+ urlparse(FIXTURE_USER_INPUT[CONF_URL])._replace(scheme="https")
)
+ for url in FIXTURE_USER_INPUT[CONF_URL], https_url:
+ requests_mock.request(ANY, url, text='')
+ requests_mock.request(
+ ANY,
+ f"{url}api/user/state-login",
+ text=(
+ f"{LoginStateEnum.LOGGED_OUT}"
+ f"{PasswordTypeEnum.SHA256}"
+ ),
+ )
+ requests_mock.request(
+ ANY,
+ f"{url}api/user/logout",
+ text="OK",
+ )
return requests_mock
@@ -194,11 +223,19 @@ async def test_login_error(
assert result["errors"] == errors
-async def test_success(hass: HomeAssistant, login_requests_mock) -> None:
+@pytest.mark.parametrize("scheme", ("http", "https"))
+async def test_success(hass: HomeAssistant, login_requests_mock, scheme: str) -> None:
"""Test successful flow provides entry creation data."""
+ user_input = {
+ **FIXTURE_USER_INPUT,
+ CONF_URL: urlunparse(
+ urlparse(FIXTURE_USER_INPUT[CONF_URL])._replace(scheme=scheme)
+ ),
+ }
+
login_requests_mock.request(
ANY,
- f"{FIXTURE_USER_INPUT[CONF_URL]}api/user/login",
+ f"{user_input[CONF_URL]}api/user/login",
text="OK",
)
with patch("homeassistant.components.huawei_lte.async_setup"), patch(
@@ -207,14 +244,14 @@ async def test_success(hass: HomeAssistant, login_requests_mock) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
- data=FIXTURE_USER_INPUT,
+ data=user_input,
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
- assert result["data"][CONF_URL] == FIXTURE_USER_INPUT[CONF_URL]
- assert result["data"][CONF_USERNAME] == FIXTURE_USER_INPUT[CONF_USERNAME]
- assert result["data"][CONF_PASSWORD] == FIXTURE_USER_INPUT[CONF_PASSWORD]
+ assert result["data"][CONF_URL] == user_input[CONF_URL]
+ assert result["data"][CONF_USERNAME] == user_input[CONF_USERNAME]
+ assert result["data"][CONF_PASSWORD] == user_input[CONF_PASSWORD]
@pytest.mark.parametrize(
@@ -300,8 +337,9 @@ async def test_ssdp(
)
for k, v in expected_result.items():
- assert result[k] == v
+ assert result[k] == v # type: ignore[literal-required] # expected is a subset
if result.get("data_schema"):
+ assert result["data_schema"] is not None
assert result["data_schema"]({})[CONF_URL] == url + "/"
@@ -355,6 +393,7 @@ async def test_reauth(
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
+ assert result["data_schema"] is not None
assert result["data_schema"]({}) == {
CONF_USERNAME: mock_entry_data[CONF_USERNAME],
CONF_PASSWORD: mock_entry_data[CONF_PASSWORD],
@@ -376,7 +415,7 @@ async def test_reauth(
await hass.async_block_till_done()
for k, v in expected_result.items():
- assert result[k] == v
+ assert result[k] == v # type: ignore[literal-required] # expected is a subset
for k, v in expected_entry_data.items():
assert entry.data[k] == v