diff --git a/gateway/platforms/bluebubbles.py b/gateway/platforms/bluebubbles.py index 909a0be66..a8a292969 100644 --- a/gateway/platforms/bluebubbles.py +++ b/gateway/platforms/bluebubbles.py @@ -224,6 +224,21 @@ class BlueBubblesAdapter(BasePlatformAdapter): host = "localhost" return f"http://{host}:{self.webhook_port}{self.webhook_path}" + @property + def _webhook_register_url(self) -> str: + """Webhook URL registered with BlueBubbles, including the password as + a query param so inbound webhook POSTs carry credentials. + + BlueBubbles posts events to the exact URL registered via + ``/api/v1/webhook``. Its webhook registration API does not support + custom headers, so embedding the password in the URL is the only + way to authenticate inbound webhooks without disabling auth. + """ + base = self._webhook_url + if self.password: + return f"{base}?password={quote(self.password, safe='')}" + return base + async def _find_registered_webhooks(self, url: str) -> list: """Return list of BB webhook entries matching *url*.""" try: @@ -245,7 +260,7 @@ class BlueBubblesAdapter(BasePlatformAdapter): if not self.client: return False - webhook_url = self._webhook_url + webhook_url = self._webhook_register_url # Crash resilience — reuse an existing registration if present existing = await self._find_registered_webhooks(webhook_url) @@ -292,7 +307,7 @@ class BlueBubblesAdapter(BasePlatformAdapter): if not self.client: return False - webhook_url = self._webhook_url + webhook_url = self._webhook_register_url removed = False try: diff --git a/tests/gateway/test_bluebubbles.py b/tests/gateway/test_bluebubbles.py index 639f81ae0..c84b1e477 100644 --- a/tests/gateway/test_bluebubbles.py +++ b/tests/gateway/test_bluebubbles.py @@ -442,6 +442,28 @@ class TestBlueBubblesWebhookUrl: adapter = _make_adapter(monkeypatch, webhook_host="192.168.1.50") assert "192.168.1.50" in adapter._webhook_url + def test_register_url_embeds_password(self, monkeypatch): + """_webhook_register_url should append ?password=... for inbound auth.""" + adapter = _make_adapter(monkeypatch, password="secret123") + assert adapter._webhook_register_url.endswith("?password=secret123") + assert adapter._webhook_register_url.startswith(adapter._webhook_url) + + def test_register_url_url_encodes_password(self, monkeypatch): + """Passwords with special characters must be URL-encoded.""" + adapter = _make_adapter(monkeypatch, password="W9fTC&L5JL*@") + assert "password=W9fTC%26L5JL%2A%40" in adapter._webhook_register_url + + def test_register_url_omits_query_when_no_password(self, monkeypatch): + """If no password is configured, the register URL should be the bare URL.""" + monkeypatch.delenv("BLUEBUBBLES_PASSWORD", raising=False) + from gateway.platforms.bluebubbles import BlueBubblesAdapter + cfg = PlatformConfig( + enabled=True, + extra={"server_url": "http://localhost:1234", "password": ""}, + ) + adapter = BlueBubblesAdapter(cfg) + assert adapter._webhook_register_url == adapter._webhook_url + class TestBlueBubblesWebhookRegistration: """Tests for _register_webhook, _unregister_webhook, _find_registered_webhooks."""