Reject LOQED webhooks without signature headers (#183877)

homeassistant/components/loqed/coordinator.py

@@ -1,11 +1,12 @@
 """Provides the coordinator for a LOQED lock."""
 
 import asyncio
+from http import HTTPStatus
 import logging
 from typing import TypedDict, override
 
 import aiohttp
-from aiohttp.web import Request
+from aiohttp.web import Request, Response
 from loqedAPI import loqed
 
 from homeassistant.components import cloud, webhook
@@ -97,9 +98,12 @@ class LoqedDataCoordinator(DataUpdateCoordinator[StatusMessage]):
 
     async def _handle_webhook(
         self, hass: HomeAssistant, webhook_id: str, request: Request
-    ) -> None:
+    ) -> Response | None:
         """Handle incoming Loqed messages."""
         _LOGGER.debug("Callback received: %s", request.headers)
+        if "TIMESTAMP" not in request.headers or "HASH" not in request.headers:
+            _LOGGER.warning("Callback without TIMESTAMP or HASH header rejected")
+            return Response(status=HTTPStatus.BAD_REQUEST)
         received_ts = request.headers["TIMESTAMP"]
         received_hash = request.headers["HASH"]
         body = await request.text()
@@ -109,9 +113,10 @@ class LoqedDataCoordinator(DataUpdateCoordinator[StatusMessage]):
         event_data = await self.lock.receiveWebhook(body, received_hash, received_ts)
         if "error" in event_data:
             _LOGGER.warning("Incorrect callback received:: %s", event_data)
-            return
+            return None
 
         self.async_update_listeners()
+        return None
 
     async def ensure_webhooks(self) -> None:
         """Register webhook on LOQED bridge."""

tests/components/loqed/test_init.py

@@ -1,6 +1,7 @@
 """Tests the init part of the Loqed integration."""
 
 from datetime import timedelta
+from http import HTTPStatus
 from typing import Any
 from unittest.mock import AsyncMock, call, patch
 
@@ -9,6 +10,7 @@ from freezegun.api import FrozenDateTimeFactory
 from loqedAPI import loqed
 import pytest
 
+from homeassistant.components.lock import LockState
 from homeassistant.components.loqed.const import DOMAIN
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.const import CONF_WEBHOOK_ID
@@ -51,6 +53,61 @@ async def test_webhook_accepts_valid_message(
     lock.receiveWebhook.assert_called()
 
 
+@pytest.mark.parametrize(
+    "headers",
+    [
+        pytest.param({"hash": "hash"}, id="missing_timestamp"),
+        pytest.param({"timestamp": "1653304609"}, id="missing_hash"),
+        pytest.param({}, id="missing_both"),
+    ],
+)
+async def test_webhook_rejects_missing_signature_headers(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    integration: MockConfigEntry,
+    lock: loqed.Lock,
+    headers: dict[str, str],
+) -> None:
+    """Test a webhook without the TIMESTAMP or HASH header is rejected."""
+    await async_setup_component(hass, "http", {"http": {}})
+    client = await hass_client_no_auth()
+    message = await async_load_fixture(hass, "battery_update.json", DOMAIN)
+
+    resp = await client.post(
+        f"/api/webhook/{integration.data[CONF_WEBHOOK_ID]}",
+        data=message,
+        headers=headers,
+    )
+
+    assert resp.status == HTTPStatus.BAD_REQUEST
+    lock.receiveWebhook.assert_not_called()
+
+
+async def test_webhook_ignores_rejected_message(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    integration: MockConfigEntry,
+    lock: loqed.Lock,
+) -> None:
+    """Test a webhook loqedAPI rejects does not update the lock state."""
+    await async_setup_component(hass, "http", {"http": {}})
+    client = await hass_client_no_auth()
+    lock.receiveWebhook = AsyncMock(return_value={"error": "Hash incorrect"})
+    lock.bolt_state = "night_lock"
+    message = await async_load_fixture(hass, "battery_update.json", DOMAIN)
+
+    resp = await client.post(
+        f"/api/webhook/{integration.data[CONF_WEBHOOK_ID]}",
+        data=message,
+        headers={"timestamp": "1653304609", "hash": "incorrect hash"},
+    )
+
+    assert resp.status == HTTPStatus.OK
+    state = hass.states.get("lock.home")
+    assert state
+    assert state.state == LockState.UNLOCKED
+
+
 async def test_setup_webhook_in_bridge(
     hass: HomeAssistant, config_entry: MockConfigEntry, lock: loqed.Lock
 ) -> None: