Fix hassio auth IndexError on Supervisor Unix socket requests

homeassistant/components/hassio/auth.py

@@ -12,6 +12,7 @@
 from homeassistant.auth.models import User
 from homeassistant.auth.providers import homeassistant as auth_ha
 from homeassistant.components.http import KEY_HASS, KEY_HASS_USER, HomeAssistantView
+from homeassistant.components.http.const import is_supervisor_unix_socket_request
 from homeassistant.components.http.data_validator import RequestDataValidator
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
@@ -41,14 +42,18 @@ def __init__(self, hass: HomeAssistant, user: User) -> None:
 
     def _check_access(self, request: web.Request) -> None:
         """Check if this call is from Supervisor."""
-        # Check caller IP
-        hassio_ip = os.environ["SUPERVISOR"].split(":")[0]
-        assert request.transport
-        if ip_address(request.transport.get_extra_info("peername")[0]) != ip_address(
-            hassio_ip
-        ):
-            _LOGGER.error("Invalid auth request from %s", request.remote)
-            raise HTTPUnauthorized
+        # Requests over the Supervisor Unix socket are authenticated by the
+        # http auth middleware as the Supervisor user, so the caller-IP check
+        # below does not apply (and would crash, since `peername` is empty for
+        # Unix sockets). The user-ID check still runs to ensure only the
+        # Supervisor user can reach this endpoint.
+        if not is_supervisor_unix_socket_request(request):
+            hassio_ip = os.environ["SUPERVISOR"].split(":")[0]
+            assert request.transport
+            peername = request.transport.get_extra_info("peername")
+            if not peername or ip_address(peername[0]) != ip_address(hassio_ip):
+                _LOGGER.error("Invalid auth request from %s", request.remote)
+                raise HTTPUnauthorized
 
         # Check caller token
         if request[KEY_HASS_USER].id != self.user.id:

tests/components/hassio/test_auth.py

@@ -1,11 +1,17 @@
 """The tests for the hassio component."""
 
 from http import HTTPStatus
-from unittest.mock import Mock, patch
+from unittest.mock import MagicMock, Mock, patch
 
 from aiohttp.test_utils import TestClient
+from aiohttp.web_exceptions import HTTPUnauthorized
+import pytest
 
 from homeassistant.auth.providers.homeassistant import InvalidAuth
+from homeassistant.components.hassio.auth import HassIOBaseAuth
+from homeassistant.components.hassio.const import DATA_CONFIG_STORE
+from homeassistant.components.http import KEY_HASS_USER
+from homeassistant.core import HomeAssistant
 
 
 async def test_auth_success(hassio_client_supervisor: TestClient) -> None:
@@ -162,6 +168,46 @@ async def test_password_fails_no_auth(hassio_noauth_client: TestClient) -> None:
     assert resp.status == HTTPStatus.UNAUTHORIZED
 
 
+@pytest.mark.parametrize(
+    ("peername", "unix_socket"),
+    [
+        # Unix socket transports report an empty string for peername. Before
+        # the fix this raised IndexError on `peername[0]`.
+        ("", True),
+        # Defensive: a TCP transport with no peername at all should be
+        # rejected, not crash.
+        (None, False),
+    ],
+)
+async def test_check_access_unix_socket_or_missing_peername(
+    hass: HomeAssistant,
+    hassio_stubs: None,
+    peername: str | None,
+    unix_socket: bool,
+) -> None:
+    """Test _check_access handles Unix socket requests and missing peername."""
+    hassio_user_id = hass.data[DATA_CONFIG_STORE].data.hassio_user
+    assert hassio_user_id is not None
+    user = await hass.auth.async_get_user(hassio_user_id)
+    assert user is not None
+
+    auth_view = HassIOBaseAuth(hass, user)
+
+    request = MagicMock()
+    request.transport.get_extra_info.return_value = peername
+    request.__getitem__.side_effect = lambda key: user if key is KEY_HASS_USER else None
+
+    with patch(
+        "homeassistant.components.hassio.auth.is_supervisor_unix_socket_request",
+        return_value=unix_socket,
+    ):
+        if unix_socket:
+            auth_view._check_access(request)
+        else:
+            with pytest.raises(HTTPUnauthorized):
+                auth_view._check_access(request)
+
+
 async def test_password_no_user(hassio_client_supervisor: TestClient) -> None:
     """Test changing password for invalid user."""
     resp = await hassio_client_supervisor.post(