Block peer certs on supervisor (#66837)

Co-authored-by: Pascal Vizeli <pvizeli@syshack.ch>
Co-authored-by: Mike Degatano <michael.degatano@gmail.com>

homeassistant/bootstrap.py

@@ -23,7 +23,7 @@ from .const import (
     SIGNAL_BOOTSTRAP_INTEGRATONS,
 )
 from .exceptions import HomeAssistantError
-from .helpers import area_registry, device_registry, entity_registry
+from .helpers import area_registry, device_registry, entity_registry, supervisor
 from .helpers.dispatcher import async_dispatcher_send
 from .helpers.typing import ConfigType
 from .setup import (
@@ -398,7 +398,7 @@ def _get_domains(hass: core.HomeAssistant, config: dict[str, Any]) -> set[str]:
         domains.update(hass.config_entries.async_domains())
 
     # Make sure the Hass.io component is loaded
-    if "HASSIO" in os.environ:
+    if supervisor.has_supervisor():
         domains.add("hassio")
 
     return domains

homeassistant/components/analytics/analytics.py

@@ -104,7 +104,7 @@ class Analytics:
     @property
     def supervisor(self) -> bool:
         """Return bool if a supervisor is present."""
-        return hassio.is_hassio(self.hass)
+        return hassio.is_hassio()
 
     async def load(self) -> None:
         """Load preferences."""

homeassistant/components/hassio/__init__.py

@@ -41,6 +41,8 @@ from homeassistant.helpers.device_registry import (
     async_get_registry,
 )
 from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.frame import report
+from homeassistant.helpers.supervisor import has_supervisor
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 from homeassistant.loader import bind_hass
@@ -394,12 +396,21 @@ def get_core_info(hass):
 
 @callback
 @bind_hass
-def is_hassio(hass: HomeAssistant) -> bool:
+def is_hassio(
+    hass: HomeAssistant | None = None,  # pylint: disable=unused-argument
+) -> bool:
     """Return true if Hass.io is loaded.
 
     Async friendly.
     """
-    return DOMAIN in hass.config.components
+    if hass is not None:
+        report(
+            "hass param deprecated for is_hassio",
+            exclude_integrations={DOMAIN},
+            error_if_core=False,
+        )
+
+    return has_supervisor()
 
 
 @callback
@@ -412,11 +423,8 @@ def get_supervisor_ip() -> str:
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:  # noqa: C901
     """Set up the Hass.io component."""
-    # Check local setup
-    for env in ("HASSIO", "HASSIO_TOKEN"):
-        if os.environ.get(env):
-            continue
-        _LOGGER.error("Missing %s environment variable", env)
+    if not has_supervisor():
+        _LOGGER.error("Supervisor not available")
         if config_entries := hass.config_entries.async_entries(DOMAIN):
             hass.async_create_task(
                 hass.config_entries.async_remove(config_entries[0].entry_id)

homeassistant/components/http/__init__.py

@@ -23,8 +23,7 @@ from homeassistant.components.network import async_get_source_ip
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP, SERVER_PORT
 from homeassistant.core import Event, HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
-from homeassistant.helpers import storage
-import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers import config_validation as cv, storage, supervisor
 from homeassistant.helpers.network import NoURLAvailableError, get_url
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.loader import bind_hass
@@ -167,6 +166,12 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     login_threshold = conf[CONF_LOGIN_ATTEMPTS_THRESHOLD]
     ssl_profile = conf[CONF_SSL_PROFILE]
 
+    if ssl_peer_certificate is not None and supervisor.has_supervisor():
+        _LOGGER.warning(
+            "Peer certificates are not supported when running the supervisor"
+        )
+        ssl_peer_certificate = None
+
     server = HomeAssistantHTTP(
         hass,
         server_host=server_host,

homeassistant/components/onboarding/views.py

@@ -195,7 +195,7 @@ class CoreConfigOnboardingView(_BaseOnboardingView):
             from homeassistant.components import hassio
 
             if (
-                hassio.is_hassio(hass)
+                hassio.is_hassio()
                 and "raspberrypi" in hassio.get_core_info(hass)["machine"]
             ):
                 onboard_integrations.append("rpi_power")

homeassistant/components/ozw/config_flow.py

@@ -45,7 +45,7 @@ class DomainConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         # Set a unique_id to make sure discovery flow is aborted on progress.
         await self.async_set_unique_id(DOMAIN, raise_on_progress=False)
 
-        if not hassio.is_hassio(self.hass):
+        if not hassio.is_hassio():
             return self._async_use_mqtt_integration()
 
         return await self.async_step_on_supervisor()

homeassistant/components/updater/__init__.py

@@ -74,7 +74,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         _LOGGER.debug("Fetched version %s: %s", newest, release_notes)
 
         # Load data from Supervisor
-        if hassio.is_hassio(hass):
+        if hassio.is_hassio():
             core_info = hassio.get_core_info(hass)
             newest = core_info["version_latest"]
 

homeassistant/components/zwave_js/config_flow.py

@@ -332,14 +332,14 @@ class ConfigFlow(BaseZwaveJSFlow, config_entries.ConfigFlow, domain=DOMAIN):
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
         """Handle the initial step."""
-        if is_hassio(self.hass):
+        if is_hassio():
             return await self.async_step_on_supervisor()
 
         return await self.async_step_manual()
 
     async def async_step_usb(self, discovery_info: usb.UsbServiceInfo) -> FlowResult:
         """Handle USB Discovery."""
-        if not is_hassio(self.hass):
+        if not is_hassio():
             return self.async_abort(reason="discovery_requires_supervisor")
         if self._async_current_entries():
             return self.async_abort(reason="already_configured")
@@ -641,7 +641,7 @@ class OptionsFlowHandler(BaseZwaveJSFlow, config_entries.OptionsFlow):
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
         """Manage the options."""
-        if is_hassio(self.hass):
+        if is_hassio():
             return await self.async_step_on_supervisor()
 
         return await self.async_step_manual()

homeassistant/helpers/supervisor.py

@@ -0,0 +1,11 @@
+"""Supervisor helper."""
+
+import os
+
+from homeassistant.core import callback
+
+
+@callback
+def has_supervisor() -> bool:
+    """Return true if supervisor is available."""
+    return "SUPERVISOR" in os.environ

tests/components/hassio/conftest.py

@@ -21,6 +21,8 @@ def hassio_env():
     ), patch.dict(os.environ, {"HASSIO_TOKEN": HASSIO_TOKEN}), patch(
         "homeassistant.components.hassio.HassIO.get_info",
         Mock(side_effect=HassioAPIError()),
+    ), patch.dict(
+        os.environ, {"SUPERVISOR": "127.0.0.1"}
     ):
         yield
 

tests/components/hassio/test_binary_sensor.py

@@ -11,7 +11,11 @@ from homeassistant.setup import async_setup_component
 
 from tests.common import MockConfigEntry
 
-MOCK_ENVIRON = {"HASSIO": "127.0.0.1", "HASSIO_TOKEN": "abcdefgh"}
+MOCK_ENVIRON = {
+    "HASSIO": "127.0.0.1",
+    "HASSIO_TOKEN": "abcdefgh",
+    "SUPERVISOR": "127.0.0.1",
+}
 
 
 @pytest.fixture(autouse=True)

tests/components/hassio/test_init.py

@@ -16,7 +16,11 @@ from homeassistant.util import dt as dt_util
 
 from tests.common import MockConfigEntry, async_fire_time_changed
 
-MOCK_ENVIRON = {"HASSIO": "127.0.0.1", "HASSIO_TOKEN": "abcdefgh"}
+MOCK_ENVIRON = {
+    "HASSIO": "127.0.0.1",
+    "HASSIO_TOKEN": "abcdefgh",
+    "SUPERVISOR": "127.0.0.1",
+}
 
 
 @pytest.fixture(autouse=True)
@@ -151,7 +155,6 @@ async def test_setup_api_ping(hass, aioclient_mock):
 
     assert aioclient_mock.call_count == 10
     assert hass.components.hassio.get_core_info()["version_latest"] == "1.0.0"
-    assert hass.components.hassio.is_hassio()
 
 
 async def test_setup_api_panel(hass, aioclient_mock):
@@ -334,7 +337,6 @@ async def test_warn_when_cannot_connect(hass, caplog):
         result = await async_setup_component(hass, "hassio", {})
         assert result
 
-    assert hass.components.hassio.is_hassio()
     assert "Not connected with the supervisor / system too busy!" in caplog.text
 
 

tests/components/hassio/test_sensor.py

@@ -11,7 +11,11 @@ from homeassistant.setup import async_setup_component
 
 from tests.common import MockConfigEntry
 
-MOCK_ENVIRON = {"HASSIO": "127.0.0.1", "HASSIO_TOKEN": "abcdefgh"}
+MOCK_ENVIRON = {
+    "HASSIO": "127.0.0.1",
+    "HASSIO_TOKEN": "abcdefgh",
+    "SUPERVISOR": "127.0.0.1",
+}
 
 
 @pytest.fixture(autouse=True)

tests/components/http/test_ban.py

@@ -36,7 +36,9 @@ def hassio_env_fixture():
     with patch.dict(os.environ, {"HASSIO": "127.0.0.1"}), patch(
         "homeassistant.components.hassio.HassIO.is_connected",
         return_value={"result": "ok", "data": {}},
-    ), patch.dict(os.environ, {"HASSIO_TOKEN": "123456"}):
+    ), patch.dict(os.environ, {"HASSIO_TOKEN": "123456"}), patch.dict(
+        os.environ, {"SUPERVISOR": "127.0.0.1"}
+    ):
         yield
 
 

tests/components/http/test_init.py

@@ -261,6 +261,42 @@ async def test_peer_cert(hass, tmpdir):
     assert len(mock_load_verify_locations.mock_calls) == 1
 
 
+async def test_peer_cert_ignored_with_supervisor(hass, tmpdir):
+    """Test peer certiicate requirement ignored in supervised deployments."""
+    cert_path, key_path, peer_cert_path = await hass.async_add_executor_job(
+        _setup_empty_ssl_pem_files, tmpdir
+    )
+
+    with patch("ssl.SSLContext.load_cert_chain"), patch(
+        "homeassistant.components.http.supervisor.has_supervisor", return_value=True
+    ), patch(
+        "ssl.SSLContext.load_verify_locations"
+    ) as mock_load_verify_locations, patch(
+        "homeassistant.util.ssl.server_context_modern",
+        side_effect=server_context_modern,
+    ) as mock_context:
+        assert (
+            await async_setup_component(
+                hass,
+                "http",
+                {
+                    "http": {
+                        "ssl_peer_certificate": peer_cert_path,
+                        "ssl_profile": "modern",
+                        "ssl_certificate": cert_path,
+                        "ssl_key": key_path,
+                    }
+                },
+            )
+            is True
+        )
+        await hass.async_start()
+        await hass.async_block_till_done()
+
+    assert len(mock_context.mock_calls) == 1
+    mock_load_verify_locations.assert_not_called()
+
+
 async def test_emergency_ssl_certificate_when_invalid(hass, tmpdir, caplog):
     """Test http can startup with an emergency self signed cert when the current one is broken."""
 

tests/components/onboarding/test_views.py

@@ -86,6 +86,8 @@ async def mock_supervisor_fixture(hass, aioclient_mock):
         return_value={"panels": {}},
     ), patch.dict(
         os.environ, {"HASSIO_TOKEN": "123456"}
+    ), patch.dict(
+        os.environ, {"SUPERVISOR": "127.0.0.1"}
     ):
         yield
 

tests/components/updater/test_init.py

@@ -7,8 +7,6 @@ from homeassistant.components import updater
 from homeassistant.helpers.update_coordinator import UpdateFailed
 from homeassistant.setup import async_setup_component
 
-from tests.common import mock_component
-
 NEW_VERSION = "10000.0"
 MOCK_VERSION = "10.0"
 MOCK_DEV_VERSION = "10.0.dev0"
@@ -113,12 +111,12 @@ async def test_new_version_shows_entity_after_hour_hassio(
     hass, mock_get_newest_version
 ):
     """Test if binary sensor gets updated if new version is available / Hass.io."""
-    mock_component(hass, "hassio")
-    hass.data["hassio_core_info"] = {"version_latest": "999.0"}
+    with patch("homeassistant.components.updater.hassio.is_hassio", return_value=True):
+        hass.data["hassio_core_info"] = {"version_latest": "999.0"}
 
-    assert await async_setup_component(hass, updater.DOMAIN, {updater.DOMAIN: {}})
+        assert await async_setup_component(hass, updater.DOMAIN, {updater.DOMAIN: {}})
 
-    await hass.async_block_till_done()
+        await hass.async_block_till_done()
 
     assert hass.states.is_state("binary_sensor.updater", "on")
     assert (

tests/helpers/test_supervisor.py

@@ -0,0 +1,16 @@
+"""Test the Hassio helper."""
+from unittest.mock import patch
+
+from homeassistant.helpers.supervisor import has_supervisor
+
+
+async def test_has_supervisor_yes():
+    """Test has_supervisor when supervisor available."""
+    with patch("homeassistant.helpers.supervisor.os.environ", {"SUPERVISOR": True}):
+        assert has_supervisor()
+
+
+async def test_has_supervisor_no():
+    """Test has_supervisor when supervisor not available."""
+    with patch("homeassistant.helpers.supervisor.os.environ"):
+        assert not has_supervisor()

tests/test_bootstrap.py

@@ -86,7 +86,7 @@ async def test_load_hassio(hass):
     with patch.dict(os.environ, {}, clear=True):
         assert bootstrap._get_domains(hass, {}) == set()
 
-    with patch.dict(os.environ, {"HASSIO": "1"}):
+    with patch.dict(os.environ, {"SUPERVISOR": "1"}):
         assert bootstrap._get_domains(hass, {}) == {"hassio"}