Check supervisor Unix socket once per request, cheapest condition first (#175568)

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Paulus Schoutsen
2026-07-04 17:33:07 -05:00
committed by GitHub
co-authored by Claude
parent fa3ededf3c
commit 7dc93c57e4
2 changed files with 74 additions and 9 deletions
+22 -9
View File
@@ -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
+52
View File
@@ -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