diff --git a/homeassistant/components/http/const.py b/homeassistant/components/http/const.py index c89751a62aff..092487101572 100644 --- a/homeassistant/components/http/const.py +++ b/homeassistant/components/http/const.py @@ -10,15 +10,28 @@ DOMAIN: Final = "http" KEY_HASS_USER: Final = "hass_user" KEY_HASS_REFRESH_TOKEN_ID: Final = "hass_refresh_token_id" +KEY_SUPERVISOR_UNIX_SOCKET: Final = "ha_supervisor_unix_socket" def is_supervisor_unix_socket_request(request: Request) -> bool: - """Check if request arrived over the Supervisor Unix socket.""" - if (transport := request.transport) is None: - return False - if (http := request.app[KEY_HASS].http) is None or ( - supervisor_path := http.supervisor_unix_socket_path - ) is None: - return False - sockname: str | None = transport.get_extra_info("sockname") - return sockname == str(supervisor_path) + """Check if request arrived over the Supervisor Unix socket. + + The result is cached on the request since it is checked by both the ban + and auth middlewares. + """ + cached: bool | None = request.get(KEY_SUPERVISOR_UNIX_SOCKET) + if cached is not None: + return cached + # Cheapest check first: without a configured socket path this can never be + # a Supervisor Unix socket request, so we avoid probing the transport. + if ( + (http := request.app[KEY_HASS].http) is None + or (supervisor_path := http.supervisor_unix_socket_path) is None + or (transport := request.transport) is None + ): + result = False + else: + sockname: str | None = transport.get_extra_info("sockname") + result = sockname == str(supervisor_path) + request[KEY_SUPERVISOR_UNIX_SOCKET] = result + return result diff --git a/tests/components/http/test_const.py b/tests/components/http/test_const.py new file mode 100644 index 000000000000..c921b18cbab2 --- /dev/null +++ b/tests/components/http/test_const.py @@ -0,0 +1,52 @@ +"""Tests for the HTTP const helpers.""" + +from pathlib import Path +from unittest.mock import MagicMock + +from aiohttp import web +from aiohttp.test_utils import make_mocked_request + +from homeassistant.components.http.const import ( + KEY_SUPERVISOR_UNIX_SOCKET, + is_supervisor_unix_socket_request, +) +from homeassistant.helpers.http import KEY_HASS + + +def _make_request(supervisor_path: Path | None, sockname: str | None) -> web.Request: + """Build a mocked request with the given supervisor socket configuration.""" + app = web.Application() + hass = MagicMock() + hass.http.supervisor_unix_socket_path = supervisor_path + app[KEY_HASS] = hass + transport = MagicMock() + transport.get_extra_info.return_value = sockname + return make_mocked_request("GET", "/", app=app, transport=transport) + + +def test_supervisor_unix_socket_request_matches() -> None: + """Test a request over the Supervisor Unix socket is detected.""" + path = Path("/run/supervisor.sock") + request = _make_request(path, str(path)) + assert is_supervisor_unix_socket_request(request) is True + + +def test_supervisor_unix_socket_request_no_path_skips_transport() -> None: + """Test the transport is not probed when no socket path is configured.""" + request = _make_request(None, "/run/supervisor.sock") + transport = request.transport + transport.get_extra_info.reset_mock() + assert is_supervisor_unix_socket_request(request) is False + transport.get_extra_info.assert_not_called() + + +def test_supervisor_unix_socket_request_is_cached() -> None: + """Test the result is computed once and cached on the request.""" + path = Path("/run/supervisor.sock") + request = _make_request(path, str(path)) + transport = request.transport + transport.get_extra_info.reset_mock() + assert is_supervisor_unix_socket_request(request) is True + assert is_supervisor_unix_socket_request(request) is True + transport.get_extra_info.assert_called_once_with("sockname") + assert request[KEY_SUPERVISOR_UNIX_SOCKET] is True