diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 307ecf46f4d7f..f05afcee44d9d 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -3457,6 +3457,8 @@ async def _stop_typing_task() -> None: if not response: logger.debug("[%s] Handler returned empty/None response for %s", self.name, event.source.chat_id) if response: + await _stop_typing_task() + # Capture [[as_document]] before extract_media strips it, so the # dispatch partition below can route image-extension files # through send_document instead of send_multiple_images. Used diff --git a/gateway/platforms/bluebubbles.py b/gateway/platforms/bluebubbles.py index 7a4af3ad68574..5d1378eabf74b 100644 --- a/gateway/platforms/bluebubbles.py +++ b/gateway/platforms/bluebubbles.py @@ -13,6 +13,7 @@ import logging import os import re +import time import uuid from datetime import datetime from typing import Any, Dict, List, Optional @@ -55,6 +56,8 @@ # Webhook event types that carry user messages _MESSAGE_EVENTS = {"new-message", "message", "updated-message"} +_INBOUND_MESSAGE_STATE_TTL_SECONDS = 300 +_INBOUND_MESSAGE_STATE_MAX_SIZE = 512 # Log redaction patterns _PHONE_RE = re.compile(r"\+?\d{7,15}") @@ -129,6 +132,8 @@ def __init__(self, config: PlatformConfig): self._private_api_enabled: Optional[bool] = None self._helper_connected: bool = False self._guid_cache: Dict[str, str] = {} + self._inbound_message_state: Dict[str, tuple[str, str, float]] = {} + self._typing_active_chats: set[str] = set() # ------------------------------------------------------------------ # API helpers @@ -249,8 +254,8 @@ async def _find_registered_webhooks(self, url: str) -> list: data = res.get("data") if isinstance(data, list): return [wh for wh in data if wh.get("url") == url] - except Exception: - pass + except Exception as exc: + logger.warning("[bluebubbles] failed to list registered webhooks: %s", exc) return [] async def _register_webhook(self) -> bool: @@ -590,6 +595,7 @@ async def send_animation( # ------------------------------------------------------------------ async def send_typing(self, chat_id: str, metadata=None) -> None: + """Show iMessage typing while Hermes is actively processing a turn.""" if not self._private_api_enabled or not self._helper_connected or not self.client: return try: @@ -599,10 +605,15 @@ async def send_typing(self, chat_id: str, metadata=None) -> None: await self.client.post( self._api_url(f"/api/v1/chat/{encoded}/typing"), timeout=5 ) - except Exception: - pass + self._typing_active_chats.add(chat_id) + except Exception as exc: + logger.warning("[bluebubbles] failed to send typing indicator: %s", exc) async def stop_typing(self, chat_id: str) -> None: + """Clear iMessage typing when Hermes finishes or is interrupted.""" + if chat_id not in self._typing_active_chats: + return + self._typing_active_chats.discard(chat_id) if not self._private_api_enabled or not self._helper_connected or not self.client: return try: @@ -612,8 +623,8 @@ async def stop_typing(self, chat_id: str) -> None: await self.client.delete( self._api_url(f"/api/v1/chat/{encoded}/typing"), timeout=5 ) - except Exception: - pass + except Exception as exc: + logger.warning("[bluebubbles] failed to stop typing indicator: %s", exc) # ------------------------------------------------------------------ # Read receipts @@ -765,6 +776,65 @@ def _value(*candidates: Any) -> Optional[str]: return candidate.strip() return None + def _prune_inbound_message_state(self, now: float) -> None: + cutoff = now - _INBOUND_MESSAGE_STATE_TTL_SECONDS + expired = [ + message_id + for message_id, (_text, _chat_id, ts) in self._inbound_message_state.items() + if ts < cutoff + ] + for message_id in expired: + self._inbound_message_state.pop(message_id, None) + while len(self._inbound_message_state) > _INBOUND_MESSAGE_STATE_MAX_SIZE: + self._inbound_message_state.pop(next(iter(self._inbound_message_state))) + + def _classify_inbound_message_update( + self, + event_type: str, + message_id: Optional[str], + text: str, + chat_id: str, + ) -> tuple[str, str, Optional[str]]: + """Classify BlueBubbles message events before dispatch. + + BlueBubbles ``updated-message`` means the macOS Messages DB row was + updated; it is not synonymous with a user edit. Delivery/read status, + chat relationship hydration, attachment metadata, and real user edits + can all arrive through the same webhook type. Hermes must therefore + compare message identity + text before creating a new user turn. + + Returns ``(kind, canonical_chat_id, previous_text)`` where kind is: + - ``new``: dispatch normally + - ``metadata``: acknowledge only; same text for an already-seen message + - ``stale_update``: acknowledge only; update arrived without prior text + - ``edit``: dispatch an explicit correction turn + """ + now = time.monotonic() + self._prune_inbound_message_state(now) + prior = self._inbound_message_state.get(message_id) if message_id else None + if event_type == "updated-message": + if not message_id: + return "stale_update", chat_id, None + if prior is None: + # BlueBubbles updates are row lifecycle events (read/delivered/edit/etc.), + # not new user turns. If Hermes missed the original new-message + # (restart or cache expiry), we cannot distinguish a real edit + # from read/delivery metadata, so acknowledge and remember this + # snapshot instead of creating a duplicate turn. + self._inbound_message_state[message_id] = (text, chat_id, now) + return "stale_update", chat_id, None + + previous_text, previous_chat_id, _previous_ts = prior + canonical_chat_id = previous_chat_id or chat_id + self._inbound_message_state[message_id] = (text, canonical_chat_id, now) + if previous_text == text: + return "metadata", canonical_chat_id, previous_text + return "edit", canonical_chat_id, previous_text + + if message_id: + self._inbound_message_state[message_id] = (text, chat_id, now) + return "new", chat_id, None + async def _handle_webhook(self, request): from aiohttp import web @@ -898,7 +968,37 @@ async def _handle_webhook(self, request): if not sender or not (chat_guid or chat_identifier) or not text: return web.json_response({"error": "missing message fields"}, status=400) - session_chat_id = chat_guid or chat_identifier + session_chat_id = str(chat_guid or chat_identifier) + message_id = self._value( + record.get("guid"), + record.get("messageGuid"), + record.get("id"), + ) + update_kind, canonical_chat_id, previous_text = self._classify_inbound_message_update( + event_type, + message_id, + text, + session_chat_id, + ) + logger.info( + "[bluebubbles] webhook message event type=%s kind=%s message=%s chat=%s chat_identifier=%s sender=%s", + event_type or "message", + update_kind, + _redact(message_id or ""), + _redact(canonical_chat_id or session_chat_id or ""), + _redact(chat_identifier or ""), + _redact(sender or ""), + ) + if update_kind in {"metadata", "stale_update"}: + return web.Response(text="ok") + if update_kind == "edit": + session_chat_id = canonical_chat_id + text = ( + "User edited a previous iMessage.\n" + f"Previous text: {previous_text or ''}\n" + f"Edited text: {text}" + ) + is_group = bool(record.get("isGroup")) or (";+;" in (chat_guid or "")) source = self.build_source( chat_id=session_chat_id, @@ -913,11 +1013,7 @@ async def _handle_webhook(self, request): message_type=msg_type, source=source, raw_message=payload, - message_id=self._value( - record.get("guid"), - record.get("messageGuid"), - record.get("id"), - ), + message_id=message_id, reply_to_message_id=self._value( record.get("threadOriginatorGuid"), record.get("associatedMessageGuid"), diff --git a/tests/gateway/test_bluebubbles.py b/tests/gateway/test_bluebubbles.py index 6f93c1d4dba64..cd32d68a00e2e 100644 --- a/tests/gateway/test_bluebubbles.py +++ b/tests/gateway/test_bluebubbles.py @@ -7,6 +7,9 @@ def _make_adapter(monkeypatch, **extra): monkeypatch.setenv("BLUEBUBBLES_SERVER_URL", "http://localhost:1234") monkeypatch.setenv("BLUEBUBBLES_PASSWORD", "secret") + monkeypatch.delenv("BLUEBUBBLES_WEBHOOK_HOST", raising=False) + monkeypatch.delenv("BLUEBUBBLES_WEBHOOK_PORT", raising=False) + monkeypatch.delenv("BLUEBUBBLES_WEBHOOK_PATH", raising=False) from gateway.platforms.bluebubbles import BlueBubblesAdapter cfg = PlatformConfig( @@ -132,6 +135,17 @@ def test_server_url_adds_scheme(self, monkeypatch): class TestBlueBubblesWebhookParsing: + class _Request: + def __init__(self, payload, password="secret"): + import json + + self.query = {"password": password} + self.headers = {} + self._body = json.dumps(payload).encode("utf-8") + + async def read(self): + return self._body + def test_webhook_prefers_chat_guid_over_message_guid(self, monkeypatch): adapter = _make_adapter(monkeypatch) payload = { @@ -273,6 +287,141 @@ def test_extract_payload_record_fallback_to_message(self, monkeypatch): record = adapter._extract_payload_record(payload) assert record["text"] == "hello" + @pytest.mark.asyncio + async def test_updated_message_metadata_update_does_not_create_second_turn(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + handled = [] + + async def fake_handle_message(event): + handled.append(event) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + base_record = { + "guid": "msg-guid-1", + "text": "hello", + "chatGuid": "iMessage;-;user@example.com", + "chatIdentifier": "user@example.com", + "handle": {"address": "user@example.com"}, + "isFromMe": False, + } + + first = await adapter._handle_webhook( + self._Request({"type": "new-message", "data": base_record}) + ) + second_record = dict(base_record) + second_record.pop("chatGuid") + second = await adapter._handle_webhook( + self._Request({"type": "updated-message", "data": second_record}) + ) + + import asyncio + + await asyncio.sleep(0) + assert first.status == 200 + assert second.status == 200 + assert len(handled) == 1 + assert handled[0].source.chat_id == "iMessage;-;user@example.com" + + @pytest.mark.asyncio + async def test_updated_message_without_prior_state_is_acknowledged_not_dispatched(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + handled = [] + + async def fake_handle_message(event): + handled.append(event) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + record = { + "guid": "msg-guid-1", + "text": "hello", + "chatIdentifier": "user@example.com", + "handle": {"address": "user@example.com"}, + "isFromMe": False, + } + + response = await adapter._handle_webhook( + self._Request({"type": "updated-message", "data": record}) + ) + + import asyncio + + await asyncio.sleep(0) + assert response.status == 200 + assert len(handled) == 0 + + @pytest.mark.asyncio + async def test_updated_message_text_change_becomes_explicit_edit_turn(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + handled = [] + + async def fake_handle_message(event): + handled.append(event) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + base_record = { + "guid": "msg-guid-1", + "text": "helo", + "chatGuid": "iMessage;-;user@example.com", + "chatIdentifier": "user@example.com", + "handle": {"address": "user@example.com"}, + "isFromMe": False, + } + + await adapter._handle_webhook( + self._Request({"type": "new-message", "data": base_record}) + ) + edited_record = dict(base_record) + edited_record["text"] = "hello" + edited_record.pop("chatGuid") + response = await adapter._handle_webhook( + self._Request({"type": "updated-message", "data": edited_record}) + ) + + import asyncio + + await asyncio.sleep(0) + assert response.status == 200 + assert len(handled) == 2 + assert handled[1].source.chat_id == "iMessage;-;user@example.com" + assert handled[1].message_id == "msg-guid-1" + assert handled[1].text == ( + "User edited a previous iMessage.\n" + "Previous text: helo\n" + "Edited text: hello" + ) + + + @pytest.mark.asyncio + async def test_typing_endpoints_send_while_processing_and_clear_afterwards(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + calls = [] + + class FakeClient: + async def post(self, url, timeout=None, **kwargs): + calls.append(("POST", url, timeout)) + + async def delete(self, url, timeout=None, **kwargs): + calls.append(("DELETE", url, timeout)) + + monkeypatch.setattr(adapter, "client", FakeClient()) + adapter._private_api_enabled = True + adapter._helper_connected = True + + async def fake_resolve(chat_id): + return "iMessage;-;user@example.com" + + monkeypatch.setattr(adapter, "_resolve_chat_guid", fake_resolve) + + await adapter.send_typing("iMessage;-;user@example.com") + await adapter.stop_typing("iMessage;-;user@example.com") + await adapter.stop_typing("iMessage;-;user@example.com") + + assert len(calls) == 2 + assert calls[0][0] == "POST" + assert "/api/v1/chat/iMessage%3B-%3Buser%40example.com/typing" in calls[0][1] + assert calls[1][0] == "DELETE" + assert "/api/v1/chat/iMessage%3B-%3Buser%40example.com/typing" in calls[1][1] + class TestBlueBubblesGuidResolution: def test_raw_guid_returned_as_is(self, monkeypatch): diff --git a/tests/tools/test_send_message_tool.py b/tests/tools/test_send_message_tool.py index 66aab5eee74c8..254bc8eb8249a 100644 --- a/tests/tools/test_send_message_tool.py +++ b/tests/tools/test_send_message_tool.py @@ -30,6 +30,8 @@ def _reset_signal_scheduler(): from tools.send_message_tool import ( _is_telegram_thread_not_found, _parse_target_ref, + _sanitize_error_text, + _send_bluebubbles, _send_matrix_via_adapter, _send_signal, _send_telegram, @@ -435,6 +437,74 @@ def _raise_and_close(coro): assert leaked not in result["error"] assert "access_token=***" in result["error"] + def test_error_sanitizer_redacts_password_query_params(self): + leaked = "bluebubbles-password-123456" + error = _sanitize_error_text( + f"403 Client Error for url 'http://localhost:1234/api?password={leaked}'" + ) + + assert leaked not in error + assert "password=***" in error + + def test_bluebubbles_send_uses_rest_without_connecting_adapter(self, monkeypatch): + import httpx + + from gateway.platforms.base import SendResult + from gateway.platforms.bluebubbles import BlueBubblesAdapter + + calls = [] + clients = [] + + class FakeAsyncClient: + def __init__(self, *args, **kwargs): + self.closed = False + clients.append(self) + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + self.closed = True + + async def fail_if_connect_called(self): + raise AssertionError("_send_bluebubbles must not start adapter.connect()") + + async def fake_api_get(self, path): + calls.append(("get", path, self.client is clients[0])) + if path == "/api/v1/server/info": + return {"data": {"private_api": True, "helper_connected": True}} + return {"data": {}} + + async def fake_send(self, chat_id, message): + calls.append(("send", chat_id, message, self._private_api_enabled, self._helper_connected)) + return SendResult(success=True, message_id="msg-1") + + monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient) + monkeypatch.setattr(BlueBubblesAdapter, "connect", fail_if_connect_called) + monkeypatch.setattr(BlueBubblesAdapter, "_api_get", fake_api_get) + monkeypatch.setattr(BlueBubblesAdapter, "send", fake_send) + + result = asyncio.run( + _send_bluebubbles( + {"server_url": "http://localhost:1234", "password": "secret"}, + "iMessage;-;user@example.com", + "hello", + ) + ) + + assert result == { + "success": True, + "platform": "bluebubbles", + "chat_id": "iMessage;-;user@example.com", + "message_id": "msg-1", + } + assert clients and clients[0].closed is True + assert calls == [ + ("get", "/api/v1/ping", True), + ("get", "/api/v1/server/info", True), + ("send", "iMessage;-;user@example.com", "hello", True, True), + ] + class TestSendTelegramMediaDelivery: def test_sends_text_then_photo_for_media_tag(self, tmp_path, monkeypatch): diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 0f83e40c3c96e..984ef5a59c396 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -50,11 +50,11 @@ # document delivery. _TELEGRAM_SEND_AUDIO_EXTS = {".mp3", ".m4a"} _URL_SECRET_QUERY_RE = re.compile( - r"([?&](?:access_token|api[_-]?key|auth[_-]?token|token|signature|sig)=)([^&#\s]+)", + r"([?&](?:access_token|api[_-]?key|auth[_-]?token|token|password|passwd|signature|sig)=)([^&#\s]+)", re.IGNORECASE, ) _GENERIC_SECRET_ASSIGN_RE = re.compile( - r"\b(access_token|api[_-]?key|auth[_-]?token|signature|sig)\s*=\s*([^\s,;]+)", + r"\b(access_token|api[_-]?key|auth[_-]?token|token|password|passwd|signature|sig)\s*=\s*([^\s,;]+)", re.IGNORECASE, ) @@ -1589,28 +1589,41 @@ async def _send_weixin(pconfig, chat_id, message, media_files=None): async def _send_bluebubbles(extra, chat_id, message): - """Send via BlueBubbles iMessage server using the adapter's REST API.""" + """Send via BlueBubbles REST without starting the inbound webhook listener.""" try: - from gateway.platforms.bluebubbles import BlueBubblesAdapter, check_bluebubbles_requirements - if not check_bluebubbles_requirements(): - return {"error": "BlueBubbles requirements not met (need aiohttp + httpx)."} + import httpx + from gateway.platforms._http_client_limits import platform_httpx_limits + from gateway.platforms.bluebubbles import BlueBubblesAdapter except ImportError: - return {"error": "BlueBubbles adapter not available."} + return {"error": "BlueBubbles adapter not available (need httpx)."} try: from gateway.config import PlatformConfig pconfig = PlatformConfig(extra=extra) adapter = BlueBubblesAdapter(pconfig) - connected = await adapter.connect() - if not connected: - return _error("BlueBubbles: failed to connect to server") + if not adapter.server_url or not adapter.password: + return _error("BlueBubbles: server URL/password not configured") + + limits = platform_httpx_limits() + if limits is None: + client = httpx.AsyncClient(timeout=30.0) + else: + client = httpx.AsyncClient(timeout=30.0, limits=limits) try: - result = await adapter.send(chat_id, message) - if not result.success: - return _error(f"BlueBubbles send failed: {result.error}") - return {"success": True, "platform": "bluebubbles", "chat_id": chat_id, "message_id": result.message_id} + async with client: + adapter.client = client + await adapter._api_get("/api/v1/ping") + info = await adapter._api_get("/api/v1/server/info") + server_data = (info or {}).get("data", {}) + adapter._private_api_enabled = bool(server_data.get("private_api")) + adapter._helper_connected = bool(server_data.get("helper_connected")) + + result = await adapter.send(chat_id, message) finally: - await adapter.disconnect() + adapter.client = None + if not result.success: + return _error(f"BlueBubbles send failed: {result.error}") + return {"success": True, "platform": "bluebubbles", "chat_id": chat_id, "message_id": result.message_id} except Exception as e: return _error(f"BlueBubbles send failed: {e}")