Add support for HTTP Digest Authentication in REST commands (#150865)

Co-authored-by: Jan-Philipp Benecke <jan-philipp@bnck.me>

homeassistant/components/rest_command/__init__.py

@@ -12,6 +12,7 @@ from aiohttp import hdrs
 import voluptuous as vol
 
 from homeassistant.const import (
+    CONF_AUTHENTICATION,
     CONF_HEADERS,
     CONF_METHOD,
     CONF_PASSWORD,
@@ -20,6 +21,8 @@ from homeassistant.const import (
     CONF_URL,
     CONF_USERNAME,
     CONF_VERIFY_SSL,
+    HTTP_BASIC_AUTHENTICATION,
+    HTTP_DIGEST_AUTHENTICATION,
     SERVICE_RELOAD,
 )
 from homeassistant.core import (
@@ -56,6 +59,9 @@ COMMAND_SCHEMA = vol.Schema(
             vol.Lower, vol.In(SUPPORT_REST_METHODS)
         ),
         vol.Optional(CONF_HEADERS): vol.Schema({cv.string: cv.template}),
+        vol.Optional(CONF_AUTHENTICATION): vol.In(
+            [HTTP_BASIC_AUTHENTICATION, HTTP_DIGEST_AUTHENTICATION]
+        ),
         vol.Inclusive(CONF_USERNAME, "authentication"): cv.string,
         vol.Inclusive(CONF_PASSWORD, "authentication"): cv.string,
         vol.Optional(CONF_PAYLOAD): cv.template,
@@ -109,10 +115,14 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         template_url = command_config[CONF_URL]
 
         auth = None
+        digest_middleware = None
         if CONF_USERNAME in command_config:
             username = command_config[CONF_USERNAME]
             password = command_config.get(CONF_PASSWORD, "")
-            auth = aiohttp.BasicAuth(username, password=password)
+            if command_config.get(CONF_AUTHENTICATION) == HTTP_DIGEST_AUTHENTICATION:
+                digest_middleware = aiohttp.DigestAuthMiddleware(username, password)
+            else:
+                auth = aiohttp.BasicAuth(username, password=password)
 
         template_payload = None
         if CONF_PAYLOAD in command_config:
@@ -155,12 +165,22 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
             )
 
             try:
+                # Prepare request kwargs
+                request_kwargs = {
+                    "data": payload,
+                    "headers": headers or None,
+                    "timeout": timeout,
+                }
+
+                # Add authentication
+                if auth is not None:
+                    request_kwargs["auth"] = auth
+                elif digest_middleware is not None:
+                    request_kwargs["middlewares"] = (digest_middleware,)
+
                 async with getattr(websession, method)(
                     request_url,
-                    data=payload,
-                    auth=auth,
-                    headers=headers or None,
-                    timeout=timeout,
+                    **request_kwargs,
                 ) as response:
                     if response.status < HTTPStatus.BAD_REQUEST:
                         _LOGGER.debug(

tests/components/rest_command/test_init.py

@@ -11,6 +11,7 @@ from homeassistant.components.rest_command import DOMAIN
 from homeassistant.const import (
     CONTENT_TYPE_JSON,
     CONTENT_TYPE_TEXT_PLAIN,
+    HTTP_DIGEST_AUTHENTICATION,
     SERVICE_RELOAD,
 )
 from homeassistant.core import HomeAssistant
@@ -123,6 +124,55 @@ async def test_rest_command_auth(
     assert len(aioclient_mock.mock_calls) == 1
 
 
+async def test_rest_command_digest_auth(
+    hass: HomeAssistant,
+    setup_component: ComponentSetup,
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Call a rest command with HTTP digest authentication."""
+    config = {
+        "digest_auth_test": {
+            "url": TEST_URL,
+            "method": "get",
+            "username": "test_user",
+            "password": "test_pass",
+            "authentication": HTTP_DIGEST_AUTHENTICATION,
+        }
+    }
+
+    await setup_component(config)
+
+    # Mock the digest auth behavior - the request will be called with DigestAuthMiddleware
+    with patch("aiohttp.ClientSession.get") as mock_get:
+
+        async def async_iter_chunks(self, chunk_size):
+            yield b"success"
+
+        mock_response = type(
+            "MockResponse",
+            (),
+            {
+                "status": 200,
+                "content_type": "text/plain",
+                "headers": {},
+                "url": TEST_URL,
+                "content": type(
+                    "MockContent", (), {"iter_chunked": async_iter_chunks}
+                )(),
+            },
+        )()
+        mock_get.return_value.__aenter__.return_value = mock_response
+
+        await hass.services.async_call(DOMAIN, "digest_auth_test", {}, blocking=True)
+
+        # Verify that the request was made with DigestAuthMiddleware
+        assert mock_get.called
+        call_kwargs = mock_get.call_args[1]
+        assert "middlewares" in call_kwargs
+        assert len(call_kwargs["middlewares"]) == 1
+        assert isinstance(call_kwargs["middlewares"][0], aiohttp.DigestAuthMiddleware)
+
+
 async def test_rest_command_form_data(
     hass: HomeAssistant,
     setup_component: ComponentSetup,