Disallow cross-origin requests for webhooks

homeassistant/components/http/__init__.py

@@ -167,6 +167,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         ssl_key=ssl_key,
         trusted_proxies=trusted_proxies,
         ssl_profile=ssl_profile,
+        cors_origins=cors_origins,
     )
     await server.async_initialize(
         cors_origins=cors_origins,
@@ -218,6 +219,7 @@ def __init__(
         server_port: int,
         trusted_proxies: list[str],
         ssl_profile: str,
+        cors_origins: list[str],
     ) -> None:
         """Initialize the HTTP Home Assistant server."""
         self.app = web.Application(middlewares=[], client_max_size=MAX_CLIENT_SIZE)
@@ -229,6 +231,7 @@ def __init__(
         self.server_port = server_port
         self.trusted_proxies = trusted_proxies
         self.ssl_profile = ssl_profile
+        self.cors_origins = cors_origins
         self.runner: web.AppRunner | None = None
         self.site: HomeAssistantTCPSite | None = None
 

homeassistant/components/webhook/__init__.py

@@ -6,7 +6,9 @@
 from ipaddress import ip_address
 import logging
 import secrets
+from urllib.parse import urlparse
 
+from aiohttp.hdrs import ORIGIN, REFERER
 from aiohttp.web import Request, Response
 import voluptuous as vol
 
@@ -106,6 +108,26 @@ async def async_handle_webhook(
         _LOGGER.debug("%s", content)
         return Response(status=HTTPStatus.OK)
 
+    # POST & HEAD are considered simple requests. They do not trigger a CORS preflight
+    # check. The presence of a "Referer" header likely indicates a CORS request. Verify
+    # that the referrer is from an allowed origin.
+    if request.headers.get(ORIGIN) is None and (
+        referrer := request.headers.get(REFERER)
+    ):
+        try:
+            url = urlparse(referrer)
+            referrer_origin = f"{url.scheme}://{url.netloc}"
+        except ValueError as err:
+            referrer_origin = f"{REFERER} header ({referrer[:16]}) does not contain a valid URL: {err}"
+
+        if referrer_origin not in hass.http.cors_origins:
+            _LOGGER.warning(
+                "Received likely CSRF request from non-trusted origin for webhook %s (%s)",
+                webhook_id,
+                referrer_origin[:80],
+            )
+            return Response(status=HTTPStatus.OK)
+
     if webhook["local_only"]:
         try:
             remote = ip_address(request.remote)
@@ -141,7 +163,6 @@ class WebhookView(HomeAssistantView):
     url = URL_WEBHOOK_PATH
     name = "api:webhook"
     requires_auth = False
-    cors_allowed = True
 
     async def _handle(self, request: Request, webhook_id: str) -> Response:
         """Handle webhook call."""

tests/components/webhook/test_init.py

@@ -3,19 +3,28 @@
 from ipaddress import ip_address
 from unittest.mock import patch
 
-from aiohttp import web
+from aiohttp import hdrs, web
 import pytest
 
 from homeassistant.components import webhook
 from homeassistant.config import async_process_ha_core_config
 from homeassistant.setup import async_setup_component
 
+TRUSTED_ORIGIN = "https://home-assistant.io"
+
 
 @pytest.fixture
-def mock_client(hass, hass_client):
+def mock_client(hass, hass_client_no_auth):
     """Create http client for webhooks."""
+    hass.loop.run_until_complete(
+        async_setup_component(
+            hass,
+            "http",
+            {"http": {"cors_allowed_origins": [TRUSTED_ORIGIN]}},
+        )
+    )
     hass.loop.run_until_complete(async_setup_component(hass, "webhook", {}))
-    return hass.loop.run_until_complete(hass_client())
+    return hass.loop.run_until_complete(hass_client_no_auth())
 
 
 async def test_unregistering_webhook(hass, mock_client):
@@ -147,6 +156,73 @@ async def handle(*args):
     assert hooks[0][2].method == "HEAD"
 
 
+async def test_webhook_cors(hass, mock_client):
+    """Test webhook with a cross-origin request."""
+    hooks = []
+    webhook_id = webhook.async_generate_id()
+
+    async def handle(*args):
+        """Handle webhook."""
+        hooks.append((args[0], args[1], await args[2].text()))
+
+    webhook.async_register(hass, "test", "Test hook", webhook_id, handle)
+
+    # Request with Referer header set from a trusted origin.
+    resp = await mock_client.post(
+        f"/api/webhook/{webhook_id}",
+        headers={"Referer": f"{TRUSTED_ORIGIN}/path?qs"},
+        json={"data": True},
+    )
+    assert resp.status == HTTPStatus.OK
+    # Should receive hook.
+    assert len(hooks) == 1
+    assert hooks[0][0] is hass
+    assert hooks[0][1] == webhook_id
+    assert hooks[0][2] == '{"data": true}'
+
+    # Request with Referer header set from a non-trusted origin.
+    resp = await mock_client.post(
+        f"/api/webhook/{webhook_id}",
+        headers={"Referer": "http://example.com/path?qs"},
+        json={"data": True},
+    )
+    assert resp.status == HTTPStatus.OK
+    # No hook received
+    assert len(hooks) == 1
+
+    # Request with invalid Referer header.
+    resp = await mock_client.post(
+        f"/api/webhook/{webhook_id}",
+        headers={"Referer": "http://[::1/128"},
+        json={"data": True},
+    )
+    assert resp.status == HTTPStatus.OK
+    # No hook received
+    assert len(hooks) == 1
+
+    # CORS preflight request for trusted origin.
+    resp = await mock_client.options(
+        f"/api/webhook/{webhook_id}",
+        headers={
+            hdrs.ORIGIN: TRUSTED_ORIGIN,
+            hdrs.ACCESS_CONTROL_REQUEST_METHOD: hdrs.METH_POST,
+        },
+        json={"data": True},
+    )
+    assert resp.status == HTTPStatus.OK
+
+    # CORS preflight request for non-trusted origin.
+    resp = await mock_client.options(
+        f"/api/webhook/{webhook_id}",
+        headers={
+            hdrs.ORIGIN: "http://example.com",
+            hdrs.ACCESS_CONTROL_REQUEST_METHOD: hdrs.METH_POST,
+        },
+        json={"data": True},
+    )
+    assert resp.status != HTTPStatus.OK
+
+
 async def test_webhook_local_only(hass, mock_client):
     """Test posting a webhook with local only."""
     hooks = []