diff --git a/homeassistant/components/loqed/coordinator.py b/homeassistant/components/loqed/coordinator.py index 20b756b3c618..f111ebc1a485 100644 --- a/homeassistant/components/loqed/coordinator.py +++ b/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.""" diff --git a/tests/components/loqed/test_init.py b/tests/components/loqed/test_init.py index 56a01873eef9..db296a6a64bc 100644 --- a/tests/components/loqed/test_init.py +++ b/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: