diff --git a/gateway/platforms/bluebubbles.py b/gateway/platforms/bluebubbles.py index 31595b223b54..62affe7c3c82 100644 --- a/gateway/platforms/bluebubbles.py +++ b/gateway/platforms/bluebubbles.py @@ -9,6 +9,7 @@ """ import asyncio +import hmac import json import logging import os @@ -864,6 +865,12 @@ def _value(*candidates: Any) -> Optional[str]: async def _handle_webhook(self, request): from aiohttp import web + # Fail closed: an unconfigured password must reject every request + # rather than authenticating callers who omit the token (``token`` is + # None/"" in that case, which would otherwise compare equal to an + # empty ``self.password``). + if not self.password: + return web.json_response({"error": "unauthorized"}, status=401) token = ( request.query.get("password") or request.query.get("guid") @@ -871,7 +878,14 @@ async def _handle_webhook(self, request): or request.headers.get("x-guid") or request.headers.get("x-bluebubbles-guid") ) - if token != self.password: + # Constant-time comparison so a mismatch can't be recovered from + # response-timing side channels. Compare UTF-8 bytes: ``hmac.compare_digest`` + # on str raises TypeError for non-ASCII code points, which would turn + # hostile/Unicode credentials into HTTP 500 instead of 401 and break + # legitimately configured non-ASCII passwords. + token_b = str(token or "").encode("utf-8") + password_b = str(self.password).encode("utf-8") + if not hmac.compare_digest(token_b, password_b): return web.json_response({"error": "unauthorized"}, status=401) try: raw = await request.read() diff --git a/tests/gateway/test_bluebubbles.py b/tests/gateway/test_bluebubbles.py index 7d4a71378c0b..c0b681914ae7 100644 --- a/tests/gateway/test_bluebubbles.py +++ b/tests/gateway/test_bluebubbles.py @@ -174,6 +174,100 @@ async def read(self): return self._body +class TestBlueBubblesWebhookAuth: + @pytest.mark.asyncio + async def test_webhook_rejects_wrong_password(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + handled = [] + + async def fake_handle_message(event): + handled.append(event) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + response = await adapter._handle_webhook( + _FakeBlueBubblesRequest({"type": "new-message"}, password="wrong") + ) + await asyncio.sleep(0) + + assert response.status == 401 + assert handled == [] + + @pytest.mark.asyncio + async def test_webhook_rejects_wrong_unicode_password(self, monkeypatch): + adapter = _make_adapter(monkeypatch, password="s3cret") + handled = [] + + async def fake_handle_message(event): + handled.append(event) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + response = await adapter._handle_webhook( + _FakeBlueBubblesRequest({"type": "new-message"}, password="пароль-неверный") + ) + await asyncio.sleep(0) + + assert response.status == 401 + assert handled == [] + + @pytest.mark.asyncio + async def test_webhook_accepts_configured_unicode_password(self, monkeypatch): + unicode_pw = "pässwörd-密钥" + adapter = _make_adapter(monkeypatch, password=unicode_pw) + handled = [] + + async def fake_handle_message(event): + handled.append(event) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + response = await adapter._handle_webhook( + _FakeBlueBubblesRequest( + { + "type": "new-message", + "data": { + "guid": "msg-unicode-auth", + "text": "hello", + "handle": {"address": "+15555550100"}, + "isFromMe": False, + "isGroup": False, + "chats": [{"guid": "iMessage;-;+15555550100"}], + }, + }, + password=unicode_pw, + ) + ) + await asyncio.sleep(0) + + assert response.status == 200 + assert len(handled) == 1 + + @pytest.mark.asyncio + async def test_connect_fails_closed_when_password_unconfigured(self, monkeypatch): + # Lifecycle invariant: connect() returns early when password is falsy + # (gateway/platforms/bluebubbles.py), so an unconfigured adapter never + # binds the aiohttp webhook route. Do not mutate adapter.password after + # construction to simulate that state — assert connect() itself refuses. + monkeypatch.setenv("BLUEBUBBLES_SERVER_URL", "http://localhost:1234") + monkeypatch.delenv("BLUEBUBBLES_PASSWORD", raising=False) + from gateway.config import PlatformConfig + from gateway.platforms.bluebubbles import BlueBubblesAdapter + + cfg = PlatformConfig( + enabled=True, + extra={ + "server_url": "http://localhost:1234", + # explicit empty password: unconfigured webhook auth + "password": "", + }, + ) + adapter = BlueBubblesAdapter(cfg) + assert not adapter.password + ok = await adapter.connect() + assert ok is False + # No webhook app should have been started for an unconfigured adapter. + assert getattr(adapter, "_runner", None) is None + assert getattr(adapter, "_site", None) is None + + class TestBlueBubblesMentionGating: @pytest.mark.asyncio async def test_group_message_without_mention_is_acknowledged_and_skipped(self, monkeypatch):