From 8dc86766f10565378481c435ddfe528927f13387 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Tue, 12 May 2026 00:48:25 +0800 Subject: [PATCH] =?UTF-8?q?feat(channels/signal):=20=E6=96=B0=E5=A2=9ESign?= =?UTF-8?q?al=E6=B8=A0=E9=81=93=E9=80=82=E9=85=8D=E5=99=A8=E5=AE=8C?= =?UTF-8?q?=E6=95=B4=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增了完整的Signal渠道适配器实现，包含RPC客户端、守护进程管理、安全策略、消息处理、安装配置工具等全套功能，支持通过signal-cli与Signal网络进行通信，包含账户管理、消息收发、反应处理、媒体分析、健康检查等能力。 --- .../yuxi/channels/adapters/signal/__init__.py | 4 + .../yuxi/channels/adapters/signal/accounts.py | 67 ++ .../yuxi/channels/adapters/signal/channel.py | 992 ++++++++++++++++++ .../yuxi/channels/adapters/signal/client.py | 187 ++++ .../channels/adapters/signal/config_schema.py | 279 +++++ .../adapters/signal/config_ui_hints.py | 269 +++++ .../yuxi/channels/adapters/signal/daemon.py | 214 ++++ .../channels/adapters/signal/directory.py | 63 ++ .../channels/adapters/signal/entity_cache.py | 68 ++ .../channels/adapters/signal/event_queue.py | 75 ++ .../channels/adapters/signal/exec_auth.py | 59 ++ .../yuxi/channels/adapters/signal/format.py | 292 ++++++ .../yuxi/channels/adapters/signal/identity.py | 41 + .../yuxi/channels/adapters/signal/install.py | 210 ++++ .../channels/adapters/signal/media_vision.py | 57 + .../adapters/signal/message_actions.py | 83 ++ .../yuxi/channels/adapters/signal/monitor.py | 241 +++++ .../channels/adapters/signal/normalize.py | 376 +++++++ .../adapters/signal/outbound_session.py | 46 + .../yuxi/channels/adapters/signal/probe.py | 137 +++ .../adapters/signal/reaction_level.py | 40 + .../channels/adapters/signal/rpc_context.py | 21 + .../yuxi/channels/adapters/signal/security.py | 214 ++++ .../yuxi/channels/adapters/signal/send.py | 428 ++++++++ .../yuxi/channels/adapters/signal/session.py | 17 + .../yuxi/channels/adapters/signal/setup.py | 180 ++++ .../channels/adapters/signal/sse_reconnect.py | 75 ++ .../channels/adapters/signal/token_utils.py | 12 + 28 files changed, 4747 insertions(+) create mode 100644 backend/package/yuxi/channels/adapters/signal/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/signal/accounts.py create mode 100644 backend/package/yuxi/channels/adapters/signal/channel.py create mode 100644 backend/package/yuxi/channels/adapters/signal/client.py create mode 100644 backend/package/yuxi/channels/adapters/signal/config_schema.py create mode 100644 backend/package/yuxi/channels/adapters/signal/config_ui_hints.py create mode 100644 backend/package/yuxi/channels/adapters/signal/daemon.py create mode 100644 backend/package/yuxi/channels/adapters/signal/directory.py create mode 100644 backend/package/yuxi/channels/adapters/signal/entity_cache.py create mode 100644 backend/package/yuxi/channels/adapters/signal/event_queue.py create mode 100644 backend/package/yuxi/channels/adapters/signal/exec_auth.py create mode 100644 backend/package/yuxi/channels/adapters/signal/format.py create mode 100644 backend/package/yuxi/channels/adapters/signal/identity.py create mode 100644 backend/package/yuxi/channels/adapters/signal/install.py create mode 100644 backend/package/yuxi/channels/adapters/signal/media_vision.py create mode 100644 backend/package/yuxi/channels/adapters/signal/message_actions.py create mode 100644 backend/package/yuxi/channels/adapters/signal/monitor.py create mode 100644 backend/package/yuxi/channels/adapters/signal/normalize.py create mode 100644 backend/package/yuxi/channels/adapters/signal/outbound_session.py create mode 100644 backend/package/yuxi/channels/adapters/signal/probe.py create mode 100644 backend/package/yuxi/channels/adapters/signal/reaction_level.py create mode 100644 backend/package/yuxi/channels/adapters/signal/rpc_context.py create mode 100644 backend/package/yuxi/channels/adapters/signal/security.py create mode 100644 backend/package/yuxi/channels/adapters/signal/send.py create mode 100644 backend/package/yuxi/channels/adapters/signal/session.py create mode 100644 backend/package/yuxi/channels/adapters/signal/setup.py create mode 100644 backend/package/yuxi/channels/adapters/signal/sse_reconnect.py create mode 100644 backend/package/yuxi/channels/adapters/signal/token_utils.py diff --git a/backend/package/yuxi/channels/adapters/signal/__init__.py b/backend/package/yuxi/channels/adapters/signal/__init__.py new file mode 100644 index 00000000..8d5678a4 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/__init__.py @@ -0,0 +1,4 @@ +from yuxi.channels.registry import register_builtin_adapter +from yuxi.channels.adapters.signal.channel import SignalChannel + +register_builtin_adapter(SignalChannel) diff --git a/backend/package/yuxi/channels/adapters/signal/accounts.py b/backend/package/yuxi/channels/adapters/signal/accounts.py new file mode 100644 index 00000000..1e67591d --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/accounts.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from typing import Any + + +def resolve_signal_account(config: dict[str, Any], account_id: str | None = None) -> dict[str, Any]: + accounts = config.get("accounts", {}) + if not accounts: + return _migrate_legacy_config(config) + + target_id = account_id or resolve_default_signal_account_id(config) + if not target_id: + raise ValueError("No Signal account configured") + + account_cfg = accounts.get(target_id) + if not account_cfg: + raise ValueError(f"Signal account '{target_id}' not found in config") + + return {**account_cfg, "_account_id": target_id} + + +def resolve_default_signal_account_id(config: dict[str, Any]) -> str | None: + accounts = config.get("accounts", {}) + if "default" in accounts: + return "default" + if accounts: + return next(iter(accounts)) + return None + + +def list_signal_account_ids(config: dict[str, Any]) -> list[str]: + accounts = config.get("accounts", {}) + if accounts: + return list(accounts) + if config.get("signal_number"): + return ["default"] + return [] + + +def list_enabled_signal_accounts(config: dict[str, Any]) -> list[dict[str, Any]]: + accounts = config.get("accounts", {}) + result = [] + for account_id, account_cfg in accounts.items(): + if account_cfg.get("enabled", True): + result.append({**account_cfg, "_account_id": account_id}) + if not result and config.get("signal_number"): + result.append(_migrate_legacy_config(config)) + return result + + +def _migrate_legacy_config(config: dict[str, Any]) -> dict[str, Any]: + return { + "signal_number": config.get("signal_number", ""), + "account_uuid": config.get("account_uuid"), + "cli_path": config.get("cli_path", "signal-cli"), + "http_host": config.get("http_host"), + "http_port": config.get("http_port"), + "http_listen": config.get("http_listen", "127.0.0.1:8080"), + "home_dir": config.get("home_dir"), + "java_opts": config.get("java_opts", "-Xmx256m"), + "receive_mode": config.get("receive_mode"), + "send_read_receipts": config.get("send_read_receipts"), + "auto_start": config.get("auto_start", True), + "enabled": config.get("enabled", True), + "startup_timeout_ms": config.get("startup_timeout_ms"), + "_account_id": "default", + } diff --git a/backend/package/yuxi/channels/adapters/signal/channel.py b/backend/package/yuxi/channels/adapters/signal/channel.py new file mode 100644 index 00000000..85b1de72 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/channel.py @@ -0,0 +1,992 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import random +import time +from collections.abc import AsyncIterator, Awaitable, Callable +from typing import Any, ClassVar + +from yuxi.channels.base import BaseChannelAdapter +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.models import ( + ChannelMessage, + ChannelResponse, + ChannelStatus, + ChannelType, + ChatType, + DeliveryResult, + HealthStatus, + MessageType, +) +from yuxi.channels.adapters.signal.accounts import ( + list_enabled_signal_accounts, +) +from yuxi.channels.adapters.signal.client import RpcClient, RpcError +from yuxi.channels.adapters.signal.daemon import SignalDaemonManager +from yuxi.channels.adapters.signal.directory import list_groups, list_peers +from yuxi.channels.adapters.signal.entity_cache import EntityCache +from yuxi.channels.adapters.signal.event_queue import OrderedEventQueue +from yuxi.channels.adapters.signal.exec_auth import ExecAuthAdapter +from yuxi.channels.adapters.signal.format import markdown_to_signal_styles +from yuxi.channels.adapters.signal.identity import get_identities +from yuxi.channels.adapters.signal.media_vision import MediaVisionAnalyzer +from yuxi.channels.adapters.signal.monitor import SSEMonitor +from yuxi.channels.adapters.signal.normalize import ( + parse_signal_delete, + parse_signal_message, + parse_signal_reaction, + parse_signal_receipt, +) +from yuxi.channels.adapters.signal.reaction_level import ReactionLevelController +from yuxi.channels.adapters.signal.send import SignalSender +from yuxi.channels.adapters.signal.security import SignalSecurityPolicy + +logger = logging.getLogger(__name__) + +STREAM_BUFFER_MAX_ENTRIES = 100 +STREAM_BUFFER_TTL = 300 +DAEMON_STARTUP_TIMEOUT_MS = 30000 + + +class SignalChannel(BaseChannelAdapter): + channel_id: ClassVar[str] = "signal" + channel_type: ClassVar[ChannelType] = ChannelType.SIGNAL + text_chunk_limit: ClassVar[int] = 4000 + supports_markdown: ClassVar[bool] = True + supports_streaming: ClassVar[bool] = True + streaming_modes: ClassVar[list[str]] = ["off", "block", "progress"] + max_media_size_mb: ClassVar[int] = 100 + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self._status = ChannelStatus.DISCONNECTED + self._daemon: SignalDaemonManager | None = None + self._rpc_client: RpcClient | None = None + self._sender: SignalSender | None = None + self._monitor: SSEMonitor | None = None + self._account_number: str = "" + self._account_uuid: str | None = None + self._supports_edit: bool = False + self._stream_buffer: dict[str, tuple[str, float]] = {} + self._security: SignalSecurityPolicy | None = None + self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60) + self._enabled = self.config.get("enabled", True) + text_chunk_limit_cfg = self.config.get("text_chunk_limit") + if text_chunk_limit_cfg is not None: + self.text_chunk_limit = int(text_chunk_limit_cfg) # type: ignore[assignment] + media_max_mb_cfg = self.config.get("media_max_mb") + if media_max_mb_cfg is not None: + self.max_media_size_mb = int(media_max_mb_cfg) # type: ignore[assignment] + coalesce_cfg = self.config.get("block_streaming_coalesce", {}) + self._stream_coalesce_min_chars: int = coalesce_cfg.get("min_chars", 1500) + self._stream_coalesce_idle_ms: int = coalesce_cfg.get("idle_ms", 1000) + reaction_level = self.config.get("reaction_level", "minimal") + self._reaction_controller = ReactionLevelController(reaction_level) + self._state_change_handlers: list[Callable[[ChannelStatus], Awaitable[None]]] = [] + self._history_store: dict[str, list[str]] = {} + self._account_instances: list[dict] = [] + + entity_cache_ttl = self.config.get("entity_cache_ttl", 600) + self._entity_cache = EntityCache(ttl=entity_cache_ttl) + self._event_queue = OrderedEventQueue( + max_size=self.config.get("event_queue_max_size", 1000), + ) + self._exec_auth = ExecAuthAdapter( + auto_approve=self.config.get("exec_auth", {}).get("auto_approve", False), + ) + self._main_dm_owner_pin = self.config.get("main_dm_owner_pin") + self._ingest_enabled = self.config.get("ingest", {}).get("enabled", False) + self._audit_enabled = self.config.get("audit", {}).get("enabled", False) + ai_vision_cfg = self.config.get("ai_vision", {}) + self._ai_vision_enabled = ai_vision_cfg.get("enabled", False) + self._lane_separator = self.config.get("lane_separator", "---") + self._warned_once: set[str] = set() + self._vision_analyzer = MediaVisionAnalyzer( + llm_call_fn=self.config.get("vision_llm_call"), + enabled=self._ai_vision_enabled, + ) + + def _warn_once(self, key: str, message: str) -> None: + if key not in self._warned_once: + self._warned_once.add(key) + logger.warning(f"[Signal] {message}") + + def on_state_change(self, handler: Callable[[ChannelStatus], Awaitable[None]]) -> None: + self._state_change_handlers.append(handler) + + async def _notify_state_change(self, new_status: ChannelStatus) -> None: + for handler in self._state_change_handlers: + try: + await handler(new_status) + except Exception: + logger.exception("State change handler failed") + + async def _on_daemon_crash(self, exit_code: int, msg: str) -> None: + logger.error(f"[Signal] Daemon crashed (exit_code={exit_code}), attempting restart") + self._status = ChannelStatus.RECONNECTING + await self._notify_state_change(ChannelStatus.RECONNECTING) + try: + if self._daemon: + await self._daemon.restart() + self._status = ChannelStatus.CONNECTED + await self._notify_state_change(ChannelStatus.CONNECTED) + except Exception: + logger.exception("[Signal] Daemon restart failed") + self._status = ChannelStatus.ERROR + await self._notify_state_change(ChannelStatus.ERROR) + + async def connect(self) -> None: + if not self._enabled: + self._status = ChannelStatus.DISABLED + logger.info(f"[Signal] Channel '{self.config.get('name', self.channel_id)}' is disabled, skipping connect") + return + + auto_start = self.config.get("auto_start", True) + if not auto_start: + logger.info(f"[Signal] auto_start is disabled for '{self.config.get('name', self.channel_id)}'") + self._status = ChannelStatus.DISABLED + return + + self._status = ChannelStatus.CONNECTING + logger.info(f"[Signal] Starting channel '{self.config.get('name', self.channel_id)}'") + + accounts = list_enabled_signal_accounts(self.config) + if not accounts: + raise ValueError("No enabled Signal accounts found in config") + + self._account_instances = [] + startup_tasks = [] + for account_cfg in accounts: + task = self._connect_account(account_cfg) + startup_tasks.append(task) + + results = await asyncio.gather(*startup_tasks, return_exceptions=True) + connected = 0 + for i, result in enumerate(results): + if isinstance(result, Exception): + logger.error(f"[Signal] Account '{accounts[i].get('_account_id')}' failed to connect: {result}") + else: + connected += 1 + + if connected == 0: + self._status = ChannelStatus.ERROR + raise RuntimeError("All Signal accounts failed to connect") + + first = self._account_instances[0] + self._daemon = first["daemon"] + self._rpc_client = first["rpc_client"] + self._sender = first["sender"] + self._monitor = first["monitor"] + self._account_number = first["number"] + self._account_uuid = first["uuid"] + self._supports_edit = first["supports_edit"] + self._security = first["security"] + + self._status = ChannelStatus.CONNECTED + logger.info(f"[Signal] Channel connected ({connected}/{len(accounts)} accounts)") + + block_streaming = self.config.get("block_streaming", False) + streaming_mode = self.config.get("streaming_mode", "block") + if block_streaming and streaming_mode != "off": + logger.warning( + f"[Signal] block_streaming=true but streaming_mode='{streaming_mode}'; " + "streaming chunks will be sent immediately without coalescing" + ) + + async def _connect_account(self, account_cfg: dict) -> None: + signal_number = account_cfg.get("signal_number", "") + if not signal_number: + raise ValueError("signal_number is required in account config") + + cli_path = account_cfg.get("cli_path", "signal-cli") + http_host = account_cfg.get("http_host") + http_port = account_cfg.get("http_port") + http_listen = account_cfg.get("http_listen") or ( + f"{http_host or '127.0.0.1'}:{http_port or 8080}" if http_host or http_port else "127.0.0.1:8080" + ) + home_dir = account_cfg.get("home_dir") + java_opts = account_cfg.get("java_opts", "-Xmx256m") + receive_mode = account_cfg.get("receive_mode") + send_read_receipts = account_cfg.get("send_read_receipts") + receive_mode_daemon_level = self.config.get("receive_mode_daemon_level", "auto") + + if receive_mode_daemon_level == "manual": + send_read_receipts = False + elif receive_mode_daemon_level == "on-start": + send_read_receipts = True + + account_uuid = account_cfg.get("account_uuid") + allow_remote_daemon = account_cfg.get("allow_remote_daemon", False) + + startup_timeout = account_cfg.get("startup_timeout_ms", DAEMON_STARTUP_TIMEOUT_MS) + daemon_startup_retries = max(1, startup_timeout // 1000) + + daemon_ready_cfg = self.config.get("daemon_ready", {}) + daemon = SignalDaemonManager( + cli_path=cli_path, + account=signal_number, + http_listen=http_listen, + home_dir=home_dir, + java_opts=java_opts, + receive_mode=receive_mode, + send_read_receipts=send_read_receipts, + daemon_startup_retries=daemon_startup_retries, + poll_interval_ms=daemon_ready_cfg.get("poll_interval_ms", 150), + log_after_ms=daemon_ready_cfg.get("log_after_ms", 10000), + log_interval_ms=daemon_ready_cfg.get("log_interval_ms", 10000), + ) + daemon.on_crash(self._on_daemon_crash) + await daemon.start() + + base_url = f"http://{daemon.listen_addr}" + http_url = self.config.get("http_url") + if http_url: + base_url = http_url.rstrip("/") + + rpc_client = RpcClient( + base_url, + timeout_ms=account_cfg.get("rpc_timeout_ms", 30000), + allow_remote_daemon=allow_remote_daemon, + ) + await rpc_client.connect() + await rpc_client.call("version") + + sender = SignalSender(rpc_client, signal_number) + supports_edit = await self._detect_edit_support_for(rpc_client) + + security_cfg = self.config.get("security", {}) + require_mention = security_cfg.get("require_mention", False) + + group_policy = security_cfg.get("group_policy") or self.config.get("group_policy") or "allowlist" + if not security_cfg.get("group_policy") and not self.config.get("group_policy"): + self._warn_once( + "missing_group_policy_fallback", + "group_policy not set in security config or channels.signal; falling back to 'allowlist'", + ) + + security = SignalSecurityPolicy( + dm_policy=security_cfg.get("dm_policy", "pairing"), + group_policy=group_policy, + allow_from=security_cfg.get("allow_from", []), + group_allow_from=security_cfg.get("group_allow_from", []), + require_mention=require_mention, + command_double_auth=self.config.get("command_double_auth", True), + ) + + monitor = SSEMonitor( + rpc_client, + signal_number, + account_uuid=account_uuid, + ignore_attachments=self.config.get("ignore_attachments", False), + ignore_stories=self.config.get("ignore_stories", False), + debounce_interval_ms=self.config.get("debounce_interval_ms", 0), + duplicate_reaction_check=self.config.get("duplicate_reaction_check", True), + event_queue=self._event_queue, + sent_message_cache=self._sent_messages, + ) + monitor.on_message(self._handle_message) + await monitor.start() + + self._account_instances.append( + { + "daemon": daemon, + "rpc_client": rpc_client, + "sender": sender, + "monitor": monitor, + "number": signal_number, + "uuid": account_uuid, + "supports_edit": supports_edit, + "security": security, + } + ) + + async def disconnect(self) -> None: + if self._status == ChannelStatus.DISCONNECTED: + return + + logger.info(f"[Signal] Stopping channel '{self.config.get('name', self.channel_id)}'") + + for instance in self._account_instances: + if instance.get("monitor"): + await instance["monitor"].stop() + if instance.get("rpc_client"): + await instance["rpc_client"].disconnect() + if instance.get("daemon"): + await instance["daemon"].stop() + + self._account_instances = [] + self._monitor = None + self._rpc_client = None + self._daemon = None + + self._stream_buffer.clear() + self._status = ChannelStatus.DISCONNECTED + + MEDIA_MESSAGE_TYPES: ClassVar[set[MessageType]] = { + MessageType.IMAGE, + MessageType.VIDEO, + MessageType.AUDIO, + MessageType.FILE, + } + + _MTYPE_TO_MEDIA: ClassVar[dict[MessageType, str]] = { + MessageType.IMAGE: "image", + MessageType.VIDEO: "video", + MessageType.AUDIO: "audio", + MessageType.FILE: "file", + } + + @staticmethod + def _media_placeholder(mime_type: str) -> str: + if mime_type.startswith("image/"): + return "" + if mime_type.startswith("video/"): + return "" + if mime_type.startswith("audio/"): + return "" + return "" + + def _resolve_media_max_bytes(self, opts_max_bytes: int | None = None) -> int: + if opts_max_bytes is not None and opts_max_bytes > 0: + return opts_max_bytes + account_cfg_max_mb = self.config.get("media_max_mb") + if account_cfg_max_mb is not None: + return int(account_cfg_max_mb) * 1024 * 1024 + return self.max_media_size_mb * 1024 * 1024 + + async def _call_with_cb(self, fn, *args, **kwargs) -> DeliveryResult: + async def _call(): + return await fn(*args, **kwargs) + + try: + return await self._circuit_breaker.call(_call) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open for Signal") + + async def send(self, response: ChannelResponse) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + + chat_id = response.identity.channel_chat_id + if not chat_id: + chat_id = self.config.get("default_to", "") + if not chat_id: + return DeliveryResult(success=False, error="No target chat_id and no default_to configured") + + async def _do_send(): + human_delay_cfg = self.config.get("human_delay", {}) + if human_delay_cfg.get("enabled", False): + min_ms = human_delay_cfg.get("min_ms", 300) + max_ms = human_delay_cfg.get("max_ms", 1500) + if max_ms > min_ms: + delay_ms = random.randint(min_ms, max_ms) + await asyncio.sleep(delay_ms / 1000.0) + + if response.message_type in self.MEDIA_MESSAGE_TYPES and response.attachments: + media_type = self._MTYPE_TO_MEDIA[response.message_type] + att = response.attachments[0] + caption = response.content or None + if caption is None and att.mime_type: + caption = self._media_placeholder(att.mime_type) + return await self.send_media( + chat_id=chat_id, + media_type=media_type, + data=att.url or b"", + filename=att.filename, + caption=caption, + ) + formatted = self._build_formatted_body(response.content) + return await self._sender.send_text( + recipient=chat_id, + message_body=response.content, + reply_to_id=response.reply_to_message_id, + formatted_body=formatted, + chunk_mode=self.config.get("chunk_mode", "newline"), + ) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open for Signal send") + + async def send_media( + self, + chat_id: str, + media_type: str, + data: Any, + filename: str | None = None, + caption: str | None = None, + ) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + + if isinstance(data, str): + import base64 as b64 + + data = b64.b64decode(data) + + if not isinstance(data, bytes): + return DeliveryResult(success=False, error="Invalid media data type") + + max_size = self._resolve_media_max_bytes() + if len(data) > max_size: + return DeliveryResult( + success=False, + error=f"Media size ({len(data)} bytes) exceeds limit ({self.max_media_size_mb}MB)", + ) + + return await self._call_with_cb( + self._sender.send_media, + recipient=chat_id, + media_data=data, + media_type=media_type, + filename=filename, + caption=caption, + ) + + async def send_reaction(self, chat_id: str, msg_id: str, emoji: str, remove: bool = False) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + + if not self._reaction_controller.should_send_reaction(chat_id, emoji): + return DeliveryResult(success=True) + + return await self._call_with_cb( + self._sender.send_reaction, + recipient=chat_id, + target_author=self._account_number, + target_sent_timestamp=int(msg_id), + reaction=emoji, + remove=remove, + ) + + async def send_read_receipt(self, chat_id: str, msg_id: str) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + + return await self._call_with_cb( + self._sender.send_read_receipt, + recipient=chat_id, + timestamps=[int(msg_id)], + ) + + async def send_chat_action(self, chat_id: str, action: str = "typing") -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + + if action == "typing": + return await self._sender.send_typing_indicator(chat_id) + return DeliveryResult(success=True) + + async def edit_message(self, chat_id: str, msg_id: str, content: str) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + if not self._supports_edit: + return DeliveryResult(success=False, error="Edit not supported by current signal-cli version") + + return await self._call_with_cb( + self._sender.edit_message, + recipient=chat_id, + target_author=self._account_number, + target_sent_timestamp=int(msg_id), + new_body=content, + ) + + async def delete_message(self, chat_id: str, msg_id: str) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + + return await self._call_with_cb( + self._sender.delete_message, + recipient=chat_id, + timestamps=[int(msg_id)], + ) + + async def send_sticker(self, chat_id: str, pack_id: str, sticker_id: int) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + return await self._call_with_cb( + self._sender.send_sticker, + recipient=chat_id, + sticker_pack_id=pack_id, + sticker_id=sticker_id, + ) + + async def send_silent_message(self, chat_id: str, content: str, reply_to_id: str | None = None) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + return await self._call_with_cb( + self._sender.send_silent_message, + recipient=chat_id, + message_body=content, + reply_to_id=reply_to_id, + ) + + async def pin_message(self, chat_id: str, msg_id: str) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + return await self._call_with_cb( + self._sender.pin_message, + recipient=chat_id, + message_timestamp=int(msg_id), + ) + + async def unpin_message(self, chat_id: str) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + return await self._call_with_cb( + self._sender.unpin_message, + recipient=chat_id, + ) + + async def send_voice(self, chat_id: str, audio_data: bytes, duration_ms: int = 0) -> DeliveryResult: + if not self._sender: + return DeliveryResult(success=False, error="Sender not initialized") + return await self._call_with_cb( + self._sender.send_voice, + recipient=chat_id, + audio_data=audio_data, + duration_ms=duration_ms, + ) + + async def send_stream_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult: + if not self._sender or not self._rpc_client: + return DeliveryResult(success=False, error="Not initialized") + + if "lane:reasoning:" in chunk: + return await self._handle_reasoning_chunk(chat_id, msg_id, chunk, finished) + + block_streaming = self.config.get("block_streaming", False) + if block_streaming: + return await self._flush_stream(chat_id, msg_id, chunk, finished) + + now = time.monotonic() + buffered, last_ts = self._stream_buffer.get(chat_id, ("", 0)) + + idle_ms = (now - last_ts) * 1000 if last_ts else 0 + if buffered and idle_ms >= self._stream_coalesce_idle_ms: + flushed = await self._flush_stream(chat_id, msg_id, buffered, False) + if not flushed.success: + return flushed + buffered = "" + + accumulated = buffered + chunk + + if finished or len(accumulated) >= self._stream_coalesce_min_chars: + self._stream_buffer.pop(chat_id, None) + return await self._flush_stream(chat_id, msg_id, accumulated, finished) + + self._stream_buffer[chat_id] = (accumulated, now) + self._prune_stream_buffer() + return DeliveryResult(success=True, message_id=msg_id) + + async def _flush_stream(self, chat_id: str, msg_id: str, content: str, finished: bool) -> DeliveryResult: + if not msg_id: + result = await self._call_with_cb(self._sender.send_text, recipient=chat_id, message_body=content) + if not finished: + self._stream_buffer[chat_id] = (content, time.monotonic()) + return result + + if not finished: + if self._supports_edit: + return await self._call_with_cb( + self._sender.edit_message, + recipient=chat_id, + target_author=self._account_number, + target_sent_timestamp=int(msg_id), + new_body=content, + ) + return DeliveryResult(success=True, message_id=msg_id) + + if self._supports_edit: + return await self._call_with_cb( + self._sender.edit_message, + recipient=chat_id, + target_author=self._account_number, + target_sent_timestamp=int(msg_id), + new_body=content, + ) + return await self._call_with_cb(self._sender.send_text, recipient=chat_id, message_body=content) + + def _prune_stream_buffer(self) -> None: + now = time.monotonic() + expired = [k for k, (_, ts) in self._stream_buffer.items() if now - ts > STREAM_BUFFER_TTL] + for k in expired: + self._stream_buffer.pop(k, None) + + if len(self._stream_buffer) > STREAM_BUFFER_MAX_ENTRIES: + sorted_keys = sorted( + self._stream_buffer.keys(), + key=lambda k: self._stream_buffer[k][1], + ) + excess = len(self._stream_buffer) - STREAM_BUFFER_MAX_ENTRIES + for k in sorted_keys[:excess]: + self._stream_buffer.pop(k, None) + + def _build_formatted_body(self, content: str) -> Any: + markdown_enabled = self.config.get("markdown_enabled", True) + if not markdown_enabled or not content: + return None + table_mode = self.config.get("markdown_table_mode", "bullets") + heading_style = self.config.get("heading_style", "bold") + blockquote_prefix = self.config.get("blockquote_prefix", "> ") + return markdown_to_signal_styles( + content, + table_mode=table_mode, + heading_style=heading_style, + blockquote_prefix=blockquote_prefix, + ) + + async def receive(self) -> AsyncIterator[ChannelMessage]: + return + yield + + def normalize_inbound(self, raw: dict) -> ChannelMessage: + for parser in (parse_signal_message, parse_signal_reaction, parse_signal_receipt, parse_signal_delete): + msg = parser(raw, channel_id=self.channel_id) + if msg is not None: + return msg + raise ValueError("Invalid signal message: no recognized event found") + + def format_outbound(self, response: ChannelResponse) -> dict: + payload: dict[str, Any] = { + "recipient": response.identity.channel_chat_id, + "messageBody": response.content, + } + if response.reply_to_message_id: + payload["quoteTimestamp"] = int(response.reply_to_message_id) + if response.message_type in self.MEDIA_MESSAGE_TYPES and response.attachments: + payload["messageBody"] = response.content or "" + payload["attachments"] = [ + { + "contentType": att.mime_type or "application/octet-stream", + "filename": att.filename or "attachment", + "id": att.file_id, + } + for att in response.attachments + ] + return payload + + async def health_check(self) -> HealthStatus: + if not self._rpc_client: + return HealthStatus(status="unhealthy", last_error="RPC client not initialized") + + try: + await self._rpc_client.call("version") + return HealthStatus( + status="healthy", + metadata={ + "account": self._account_number, + "stream_edit_support": self._supports_edit, + "adapter_status": self._status.value, + }, + ) + except Exception as e: + return HealthStatus(status="unhealthy", last_error=str(e)) + + async def _handle_message(self, message: ChannelMessage) -> None: + security_reason = None + + if self._security is not None: + if message.chat_type == ChatType.GROUP: + if not self._security.check_group_permission(message): + logger.debug(f"[Signal] Group message blocked by security: {message.identity.channel_chat_id}") + if self._audit_enabled: + self._log_audit(message, "denied", "group_policy") + if self._ingest_enabled: + await self._ingest_group_message(message) + return + if not self._security.check_require_mention(message): + if self._audit_enabled: + self._log_audit(message, "filtered", "require_mention") + if self._ingest_enabled: + await self._ingest_group_message(message) + return + if self._security.require_mention: + if message.mentions and message.mentions.is_bot_mentioned: + if not self._security.check_command_double_auth(message): + logger.info(f"[Signal] Command double-auth failed for {message.identity.channel_user_id}") + if self._audit_enabled: + self._log_audit(message, "denied", "command_double_auth") + return + else: + if not self._security.check_dm_permission(message): + logger.debug(f"[Signal] DM blocked by security: {message.identity.channel_user_id}") + if self._audit_enabled: + self._log_audit(message, "denied", "dm_policy") + return + if self._main_dm_owner_pin and message.identity.channel_user_id != self._main_dm_owner_pin: + logger.debug( + f"[Signal] DM skipped by ownerPin: expected={self._main_dm_owner_pin}, " + f"got={message.identity.channel_user_id}" + ) + return + + if not self._check_context_visibility(message): + return + + content_lower = (message.content or "").lower() + if not self.allow_config_writes() and self._is_config_command(content_lower): + logger.info(f"[Signal] Config write denied (config_writes=false): {message.identity.channel_user_id}") + return + + if not self._exec_auth.auto_approve and self._is_sensitive_command(content_lower): + approved = await self._send_with_exec_auth(message.identity.channel_chat_id, message) + if not approved: + if self._audit_enabled: + self._log_audit(message, "denied", "exec_auth") + return + + if self._audit_enabled: + self._log_audit(message, "allowed", security_reason) + + if self._ai_vision_enabled and message.message_type == MessageType.IMAGE and message.attachments: + await self._analyze_vision(message) + + if self._reaction_controller.should_send_auto_ack(message.identity.channel_chat_id): + await self.send_reaction( + message.identity.channel_chat_id, + message.identity.channel_message_id, + "\U0001f440", + ) + + history_limit = self.config.get("history_limit") + if history_limit is not None and message.chat_type == ChatType.GROUP: + chat_id = message.identity.channel_chat_id + if chat_id not in self._history_store: + self._history_store[chat_id] = [] + recent = self._history_store[chat_id] + msg_id = message.identity.channel_message_id + if msg_id in recent: + return + recent.append(msg_id) + if len(recent) > history_limit: + self._history_store[chat_id] = recent[-history_limit:] + + await super()._handle_message(message) + + async def _ingest_group_message(self, message: ChannelMessage) -> None: + sender = message.identity.channel_user_id + group = message.identity.channel_chat_id + content_preview = message.content[:200] if message.content else "(no text)" + logger.info(f"[Signal] Group silent ingest: sender={sender}, group={group}, content_preview={content_preview}") + + @staticmethod + def _is_config_command(content: str) -> bool: + config_keywords = ["/config", "/set ", "/settings", "config_writes", "dm_policy", "group_policy"] + return any(kw in content for kw in config_keywords) + + @staticmethod + def _is_sensitive_command(content: str) -> bool: + sensitive_keywords = ["/delete", "/edit", "/unsend", "/send_media", "/media"] + return any(kw in content for kw in sensitive_keywords) + + async def _analyze_vision(self, message: ChannelMessage) -> None: + for att in message.attachments: + if not att.mime_type or not att.mime_type.startswith("image/"): + continue + try: + if att.file_id: + image_data = await self.download_media(att.file_id) + description = await self._vision_analyzer.analyze_image(image_data, att.mime_type) + if description: + existing = message.content or "" + prefix = f"{existing}\n[image: {description}]" if existing else f"[image: {description}]" + message.content = prefix + except Exception: + logger.exception("Vision analysis failed for incoming message") + + def _log_audit(self, message: ChannelMessage, decision: str, reason: str | None = None) -> None: + entry = { + "timestamp": int(time.time() * 1000), + "channel": "signal", + "user_id": message.identity.channel_user_id, + "chat_id": message.identity.channel_chat_id, + "chat_type": message.chat_type.value, + "message_id": message.identity.channel_message_id, + "action": message.event_type.value, + "decision": decision, + "reason": reason or "", + "dm_policy": str(self._security.dm_policy) if self._security else "none", + "group_policy": str(self._security.group_policy) if self._security else "none", + } + logger.info(f"[Signal Audit] {json.dumps(entry, ensure_ascii=False)}") + + def _check_context_visibility(self, message: ChannelMessage) -> bool: + mode = self.config.get("context_visibility_mode", "all") + if mode == "all": + return True + if mode == "none": + return False + if mode == "same-group": + return message.chat_type == ChatType.GROUP + if mode == "trusted": + user_id = message.identity.channel_user_id + return self._security is not None and self._security.check_dm_permission(message) + return True + + async def get_user_info(self, channel_user_id: str) -> dict[str, Any]: + if not self._rpc_client: + return {} + + cached = await self._entity_cache.get(f"user:{channel_user_id}") + if cached is not None: + return cached + + try: + identities = await get_identities(self._rpc_client, self._account_number) + for identity in identities: + if identity.get("number") == channel_user_id: + info = { + "number": identity.get("number", channel_user_id), + "name": identity.get("name", ""), + "fingerprint": identity.get("fingerprint", ""), + "trust_level": identity.get("trustLevel", "UNTRUSTED"), + "added": identity.get("added", ""), + } + await self._entity_cache.set(f"user:{channel_user_id}", info) + return info + info = {"number": channel_user_id} + await self._entity_cache.set(f"user:{channel_user_id}", info) + return info + except RpcError as e: + logger.warning(f"get_user_info failed for {channel_user_id}: {e}") + return {} + + async def list_peers(self) -> list[dict]: + if not self._rpc_client: + return [] + + cached = await self._entity_cache.get("peers:list") + if cached is not None: + return cached + + peers = await list_peers(self._rpc_client, self._account_number) + await self._entity_cache.set("peers:list", peers) + return peers + + async def list_groups(self) -> list[dict]: + if not self._rpc_client: + return [] + + cached = await self._entity_cache.get("groups:list") + if cached is not None: + return cached + + groups = await list_groups(self._rpc_client, self._account_number) + await self._entity_cache.set("groups:list", groups) + return groups + + async def download_media(self, file_id: str) -> bytes: + if not self._rpc_client: + raise RuntimeError("RPC client not initialized") + + max_size_mb = self.config.get("attachment_max_size_mb", 0) + if max_size_mb > 0: + max_size_bytes = max_size_mb * 1024 * 1024 + + try: + result = await self._rpc_client.call( + "getAttachment", + {"account": self._account_number, "attachmentId": file_id}, + ) + data_b64 = result.get("data", "") + import base64 as b64 + + data = b64.b64decode(data_b64) + if max_size_mb > 0 and len(data) > max_size_bytes: + raise RuntimeError(f"Attachment size ({len(data)} bytes) exceeds limit ({max_size_mb}MB)") + return data + except RpcError as e: + raise RuntimeError(f"Failed to download attachment {file_id}: {e}") + + async def pre_connect(self) -> dict: + signal_number = self.config.get("signal_number", "") + if not signal_number: + return {"status": "error", "message": "Missing signal_number"} + + cli_path = self.config.get("cli_path", "signal-cli") + import shutil + + if not shutil.which(cli_path) and cli_path != "signal-cli": + return {"status": "error", "message": f"signal-cli not found at: {cli_path}"} + + http_listen = self.config.get("http_listen", "127.0.0.1:8080") + + rpc_client = RpcClient(f"http://{http_listen}") + try: + await rpc_client.connect() + result = await rpc_client.call("version") + return { + "status": "ok", + "version": result.get("version", "unknown"), + "account": signal_number, + } + except Exception as e: + return {"status": "error", "message": str(e)} + finally: + try: + await rpc_client.disconnect() + except Exception: + pass + + @property + def base_url(self) -> str: + return self._rpc_client.base_url if self._rpc_client else "" + + def allow_config_writes(self) -> bool: + return self.config.get("config_writes", False) + + def get_sent_message_status(self, msg_id: str) -> dict | None: + if not self._sender: + return None + return self._sender._get_cached(msg_id) + + async def _detect_edit_support(self) -> bool: + if not self._rpc_client: + return False + return await self._detect_edit_support_for(self._rpc_client) + + async def _detect_edit_support_for(self, rpc_client: RpcClient) -> bool: + edit_mode = self.config.get("streaming", {}).get("edit_support", "auto") + if edit_mode == "disabled": + return False + if edit_mode == "force": + return True + + try: + version_result = await rpc_client.call("version") + version_str = version_result.get("version", "0.0") + parts = version_str.lstrip("v").split(".") + major = int(parts[0]) if parts else 0 + minor = int(parts[1]) if len(parts) > 1 else 0 + return (major > 0) or (major == 0 and minor >= 13) + except Exception: + return False + + async def _handle_reasoning_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult: + reasoning_content = chunk.replace("lane:reasoning:", "") + sep = self._lane_separator + prefix = f"(thinking)\n{reasoning_content}\n{sep}\n" + return await self._flush_stream(chat_id, msg_id, prefix, finished) + + async def _send_with_exec_auth(self, chat_id: str, message: ChannelMessage) -> bool: + action = message.metadata.get("command") or message.event_type.value + user_id = message.identity.channel_user_id + params = { + "content": message.content, + "chat_id": chat_id, + "message_type": message.message_type.value, + } + from yuxi.channels.adapters.signal.exec_auth import ExecAuthResult + + result = await self._exec_auth.request_approval(action, params, user_id) + if result == ExecAuthResult.DENIED: + logger.info(f"[Signal] Exec auth denied for {action} by {user_id}") + return False + if result == ExecAuthResult.PENDING: + logger.info(f"[Signal] Exec auth pending for {action} by {user_id}") + return False + return True diff --git a/backend/package/yuxi/channels/adapters/signal/client.py b/backend/package/yuxi/channels/adapters/signal/client.py new file mode 100644 index 00000000..1438f34e --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/client.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +import asyncio +import ipaddress +import logging +import random +from urllib.parse import urlparse + +import aiohttp + +logger = logging.getLogger(__name__) + +LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1", "0.0.0.0"} +PRIVATE_NETWORKS = [ + ipaddress.IPv4Network("10.0.0.0/8"), + ipaddress.IPv4Network("172.16.0.0/12"), + ipaddress.IPv4Network("192.168.0.0/16"), +] + + +class RpcError(Exception): + def __init__(self, message: str, code: int = -1): + super().__init__(message) + self.code = code + + +class RateLimitError(RpcError): + def __init__(self, message: str, retry_after_seconds: int | None = None, token: str | None = None): + super().__init__(message, code=-32603) + self.retry_after_seconds = retry_after_seconds + self.token = token + + +def _detect_rate_limit(message: str, error_data: dict) -> RateLimitError | None: + msg_lower = message.lower() + if "rate limit" in msg_lower or "ratelimitexception" in msg_lower: + retry_after = error_data.get("retry_after_seconds") + token = error_data.get("token") + return RateLimitError(message, retry_after_seconds=retry_after, token=token) + return None + + +RATE_LIMIT_CODE = -32603 + + +class UnauthorizedError(RpcError): + def __init__(self, message: str = "Unauthorized"): + super().__init__(message, code=-32001) + + +class RpcClient: + DEFAULT_RETRY = { + "attempts": 3, + "min_delay_ms": 400, + "max_delay_ms": 30000, + "jitter": 0.1, + } + + def __init__( + self, + base_url: str, + retry_config: dict | None = None, + timeout_ms: int = 30000, + allow_remote_daemon: bool = False, + ): + self._base_url = base_url.rstrip("/") + self._session: aiohttp.ClientSession | None = None + self._request_id = 0 + self._retry_config = {**self.DEFAULT_RETRY, **(retry_config or {})} + self._timeout_ms = timeout_ms + self._allow_remote_daemon = allow_remote_daemon + + async def connect(self) -> None: + if not self._allow_remote_daemon: + _validate_daemon_url(self._base_url) + self._session = aiohttp.ClientSession( + base_url=self._base_url, + timeout=aiohttp.ClientTimeout(total=self._timeout_ms / 1000.0), + ) + + async def disconnect(self) -> None: + if self._session: + await self._session.close() + self._session = None + + async def call(self, method: str, params: dict | None = None) -> dict: + self._request_id += 1 + payload = { + "jsonrpc": "2.0", + "method": method, + "params": params or {}, + "id": str(self._request_id), + } + + if not self._session: + raise RuntimeError("RpcClient not connected") + + attempts = self._retry_config["attempts"] + max_delay = self._retry_config["max_delay_ms"] / 1000.0 + jitter = self._retry_config["jitter"] + last_error: Exception | None = None + + for attempt in range(attempts): + try: + return await self._do_call(payload) + except (aiohttp.ClientError, TimeoutError) as e: + last_error = e + if attempt < attempts - 1: + delay = min( + self._retry_config["min_delay_ms"] / 1000.0 * (2**attempt), + max_delay, + ) + delay += delay * jitter * random.random() + await asyncio.sleep(delay) + except RpcError: + raise + except UnauthorizedError: + raise + + raise last_error # type: ignore[misc] + + async def _do_call(self, payload: dict) -> dict: + async with self._session.post("/api/v1/rpc", json=payload) as resp: + if resp.status == 401: + raise UnauthorizedError("HTTP 401 Unauthorized") + + result = await resp.json() + + if "error" in result: + err = result["error"] + message = err.get("message", "Unknown RPC error") + code = err.get("code", -1) + + rate_limit = _detect_rate_limit(message, err) + if rate_limit: + logger.warning(f"Rate limit detected: {rate_limit}") + raise rate_limit + + raise RpcError(f"RPC error [{code}]: {message}", code=code) + + return result.get("result", {}) + + async def multipart_upload(self, endpoint: str, data: aiohttp.FormData) -> dict: + if not self._session: + raise RuntimeError("RpcClient not connected") + + async with self._session.post(endpoint, data=data) as resp: + return await resp.json() + + @property + def base_url(self) -> str: + return self._base_url + + +def _validate_daemon_url(base_url: str) -> None: + parsed = urlparse(base_url) + hostname = parsed.hostname + if not hostname: + raise ValueError(f"Invalid daemon URL: cannot determine host from {base_url}") + + if hostname in LOOPBACK_HOSTS: + return + + try: + addr = ipaddress.IPv4Address(hostname) + except ValueError: + try: + addr = ipaddress.IPv6Address(hostname) + except ValueError: + if hostname not in LOOPBACK_HOSTS: + raise ValueError( + f"Remote daemon URL rejected (SSRF guard): {base_url}. " + "Set allow_remote_daemon=True to permit non-local connections." + ) + return + + if addr.is_loopback: + return + + for network in PRIVATE_NETWORKS: + if isinstance(addr, ipaddress.IPv4Address) and addr in network: + return + + raise ValueError( + f"Remote daemon URL rejected (SSRF guard): {base_url}. " + "Set allow_remote_daemon=True to permit non-local connections." + ) diff --git a/backend/package/yuxi/channels/adapters/signal/config_schema.py b/backend/package/yuxi/channels/adapters/signal/config_schema.py new file mode 100644 index 00000000..ee414c36 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/config_schema.py @@ -0,0 +1,279 @@ +from __future__ import annotations + +SIGNAL_CONFIG_SCHEMA = { + "enabled": { + "type": "bool", + "default": True, + "description": "Whether the Signal channel is enabled", + }, + "signal_number": { + "type": "str", + "default": "", + "description": "Signal E.164 phone number (legacy single-account config)", + }, + "account_uuid": { + "type": "str", + "default": None, + "description": "Signal account UUID for self-loop detection", + }, + "accounts": { + "type": "dict", + "default": {}, + "description": "Multi-account configuration map (account_id -> config)", + }, + "cli_path": { + "type": "str", + "default": "signal-cli", + "description": "Path to signal-cli binary", + }, + "http_host": { + "type": "str", + "default": "127.0.0.1", + "description": "HTTP daemon listen host", + }, + "http_port": { + "type": "int", + "default": 8080, + "description": "HTTP daemon listen port", + }, + "http_listen": { + "type": "str", + "default": "127.0.0.1:8080", + "description": "HTTP daemon listen address (host:port, overrides http_host/http_port)", + }, + "http_url": { + "type": "str", + "default": None, + "description": "Custom HTTP URL (overrides host/port)", + }, + "home_dir": { + "type": "str", + "default": None, + "description": "signal-cli data directory", + }, + "java_opts": { + "type": "str", + "default": "-Xmx256m", + "description": "Java runtime options", + }, + "receive_mode": { + "type": "str", + "default": None, + "description": "Daemon receive mode: 'on-start' or 'manual'", + }, + "send_read_receipts": { + "type": "bool", + "default": None, + "description": "Whether daemon should send read receipts", + }, + "auto_start": { + "type": "bool", + "default": True, + "description": "Whether to auto-start the daemon", + }, + "startup_timeout_ms": { + "type": "int", + "default": 30000, + "description": "Daemon startup timeout in milliseconds", + }, + "rpc_timeout_ms": { + "type": "int", + "default": 30000, + "description": "RPC call timeout in milliseconds", + }, + "media_max_mb": { + "type": "int", + "default": 8, + "description": "Maximum media file size in MB", + }, + "text_chunk_limit": { + "type": "int", + "default": 4000, + "description": "Maximum text chunk length", + }, + "chunk_mode": { + "type": "str", + "default": "newline", + "description": "Text chunking mode: 'length' or 'newline'", + }, + "markdown_enabled": { + "type": "bool", + "default": True, + "description": "Whether to convert Markdown to Signal rich text", + }, + "markdown_table_mode": { + "type": "str", + "default": "bullets", + "description": "Markdown table conversion mode", + }, + "heading_style": { + "type": "str", + "default": "bold", + "description": "Heading conversion style", + }, + "blockquote_prefix": { + "type": "str", + "default": "> ", + "description": "Blockquote prefix", + }, + "reaction_level": { + "type": "str", + "default": "minimal", + "description": "Reaction level: 'off', 'ack', 'minimal', or 'extensive'", + }, + "reaction_notifications": { + "type": "str", + "default": "all", + "description": "Reaction notification policy: 'off', 'own', 'allowlist', or 'all'", + }, + "reaction_allowlist": { + "type": "list", + "default": [], + "description": "Reaction notification allowlist", + }, + "history_limit": { + "type": "int", + "default": 50, + "description": "Group chat history context limit", + }, + "block_streaming": { + "type": "bool", + "default": False, + "description": "Whether to disable block streaming mode", + }, + "block_streaming_coalesce": { + "type": "dict", + "default": {"min_chars": 1500, "idle_ms": 1000}, + "description": "Streaming coalesce parameters", + }, + "ignore_attachments": { + "type": "bool", + "default": False, + "description": "Whether to ignore incoming attachments", + }, + "ignore_stories": { + "type": "bool", + "default": False, + "description": "Whether to ignore stories", + }, + "config_writes": { + "type": "bool", + "default": False, + "description": "Whether to allow config writes via Signal messages", + }, + "name": { + "type": "str", + "default": "", + "description": "Account display name", + }, + "default_to": { + "type": "str", + "default": None, + "description": "Default send target", + }, + "context_visibility_mode": { + "type": "str", + "default": "all", + "description": "Context visibility: 'all', 'same-group', 'trusted', or 'none'", + }, + "security": { + "type": "dict", + "default": { + "dm_policy": "pairing", + "group_policy": "allowlist", + "allow_from": [], + "group_allow_from": [], + "require_mention": False, + }, + "description": "Security policy config", + }, + "streaming": { + "type": "dict", + "default": {"edit_support": "auto"}, + "description": "Streaming config", + }, + "allow_remote_daemon": { + "type": "bool", + "default": False, + "description": "Allow non-loopback daemon URL (disable SSRF guard)", + }, + "download_max_mb": { + "type": "int", + "default": 256, + "description": "Maximum signal-cli download size in MB", + }, + "entity_cache_ttl": { + "type": "int", + "default": 600, + "description": "Entity cache TTL in seconds", + }, + "event_queue_max_size": { + "type": "int", + "default": 1000, + "description": "Maximum event queue size", + }, + "prefer_native_binary": { + "type": "bool", + "default": True, + "description": "Prefer GraalVM native binary over universal tar.gz", + }, + "receive_mode_daemon_level": { + "type": "str", + "default": "auto", + "description": "Daemon-level receipt mode: 'auto', 'on-start', or 'manual'", + }, + "human_delay": { + "type": "dict", + "default": {"enabled": False, "min_ms": 300, "max_ms": 1500}, + "description": "Human typing delay simulation config", + }, + "ingest": { + "type": "dict", + "default": {"enabled": False}, + "description": "Group silent ingest config (fire-and-forget hook for skipped messages)", + }, + "main_dm_owner_pin": { + "type": "str", + "default": None, + "description": "Pin DM routing to a specific allowlisted sender (E.164/UUID)", + }, + "daemon_ready": { + "type": "dict", + "default": { + "poll_interval_ms": 150, + "log_after_ms": 10000, + "log_interval_ms": 10000, + }, + "description": "Daemon readiness check timing parameters", + }, + "duplicate_reaction_check": { + "type": "bool", + "default": True, + "description": "Enable composite-key reaction deduplication", + }, + "command_double_auth": { + "type": "bool", + "default": True, + "description": "Enable dual-verifier (DM allowFrom + group allowFrom) for commands", + }, + "pairing_store_persist": { + "type": "bool", + "default": True, + "description": "Persist pairing approvals to Store for cross-restart survival", + }, + "audit": { + "type": "dict", + "default": {"enabled": False}, + "description": "Structured security audit logging config", + }, + "ai_vision": { + "type": "dict", + "default": {"enabled": False}, + "description": "AI-powered media vision analysis config", + }, + "lane_separator": { + "type": "str", + "default": "---", + "description": "Reasoning lane separator for streaming output", + }, +} diff --git a/backend/package/yuxi/channels/adapters/signal/config_ui_hints.py b/backend/package/yuxi/channels/adapters/signal/config_ui_hints.py new file mode 100644 index 00000000..2b16d1ab --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/config_ui_hints.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +SIGNAL_CONFIG_UI_HINTS = { + "enabled": { + "label": "Enable Signal Channel", + "help_text": "Toggle this channel on/off without removing its configuration.", + "group": "general", + }, + "signal_number": { + "label": "Signal Phone Number", + "help_text": "E.164 format phone number (e.g. +8613800138000). Used as fallback for single-account configs.", + "group": "account", + }, + "account_uuid": { + "label": "Account UUID", + "help_text": "Signal account UUID for self-loop detection. Leave blank to auto-detect on first connect.", + "group": "account", + }, + "accounts": { + "label": "Multi-Account Config", + "help_text": "Map of account_id → account config. Each account gets its own daemon + monitor instance.", + "group": "account", + }, + "cli_path": { + "label": "signal-cli Binary Path", + "help_text": "Path to signal-cli binary. Default 'signal-cli' searches PATH.", + "group": "daemon", + }, + "http_host": { + "label": "Daemon Host", + "help_text": "HTTP daemon listen host. Default: 127.0.0.1 (loopback only for security).", + "group": "daemon", + }, + "http_port": { + "label": "Daemon Port", + "help_text": "HTTP daemon listen port. Default: 8080.", + "group": "daemon", + }, + "http_listen": { + "label": "Daemon Listen Address", + "help_text": "Combined host:port override for http_host and http_port. Default: 127.0.0.1:8080.", + "group": "daemon", + }, + "http_url": { + "label": "Custom HTTP URL", + "help_text": "Override the full daemon URL. Useful for remote daemons (requires allow_remote_daemon=true).", + "group": "daemon", + }, + "home_dir": { + "label": "signal-cli Data Directory", + "help_text": "Custom signal-cli data directory. Leave empty for default (~/.local/share/signal-cli).", + "group": "daemon", + }, + "java_opts": { + "label": "Java Options", + "help_text": "JVM options passed to signal-cli. Default: -Xmx256m.", + "group": "daemon", + }, + "receive_mode": { + "label": "Daemon Receive Mode", + "help_text": "How the daemon receives messages: 'on-start' (fetch on startup) or 'manual' (app controls).", + "group": "daemon", + }, + "send_read_receipts": { + "label": "Daemon Read Receipts", + "help_text": "Let the daemon auto-send read receipts. Set to false for program-controlled receipts.", + "group": "daemon", + }, + "auto_start": { + "label": "Auto-Start Daemon", + "help_text": "Automatically start the signal-cli daemon when the channel connects.", + "group": "daemon", + }, + "startup_timeout_ms": { + "label": "Daemon Startup Timeout (ms)", + "help_text": "Maximum time to wait for the daemon to become ready. Default: 30000ms.", + "group": "daemon", + }, + "rpc_timeout_ms": { + "label": "RPC Call Timeout (ms)", + "help_text": "Timeout for individual JSON-RPC calls to signal-cli. Default: 30000ms.", + "group": "rpc", + }, + "media_max_mb": { + "label": "Max Media Size (MB)", + "help_text": "Maximum allowed media file size in megabytes. Files exceeding this are rejected.", + "group": "messaging", + }, + "text_chunk_limit": { + "label": "Text Chunk Limit", + "help_text": "Maximum characters per message chunk. Longer messages are split. Default: 4000.", + "group": "messaging", + }, + "chunk_mode": { + "label": "Chunk Splitting Mode", + "help_text": "How to split long messages: 'newline' (at line boundaries) or 'length' (by character count).", + "group": "messaging", + }, + "markdown_enabled": { + "label": "Enable Markdown", + "help_text": "Convert Markdown formatting to Signal rich text styles (bold, italic, strikethrough, etc.).", + "group": "messaging", + }, + "markdown_table_mode": { + "label": "Table Rendering Mode", + "help_text": "How to convert Markdown tables: 'bullets' (list items) or pass-through.", + "group": "messaging", + }, + "heading_style": { + "label": "Heading Style", + "help_text": "How to render Markdown headings: 'bold' or plain text.", + "group": "messaging", + }, + "blockquote_prefix": { + "label": "Blockquote Prefix", + "help_text": "Character prefix for blockquote lines. Default: '> '.", + "group": "messaging", + }, + "reaction_level": { + "label": "Reaction Level", + "help_text": "Auto-reaction control: 'off' (none), 'ack' (only 👀), 'minimal' (basic), 'extensive' (all).", + "group": "reactions", + }, + "reaction_notifications": { + "label": "Reaction Notifications", + "help_text": "Who can trigger reaction notifications: 'off', 'own', 'allowlist', or 'all'.", + "group": "reactions", + }, + "reaction_allowlist": { + "label": "Reaction Allowlist", + "help_text": "List of user IDs allowed to trigger reaction notifications.", + "group": "reactions", + }, + "history_limit": { + "label": "Group History Limit", + "help_text": "Maximum messages to keep per group for deduplication. Default: 50.", + "group": "processing", + }, + "block_streaming": { + "label": "Block Streaming Mode", + "help_text": "When enabled, streaming chunks are sent immediately without coalescing (progress mode).", + "group": "streaming", + }, + "block_streaming_coalesce": { + "label": "Streaming Coalesce Params", + "help_text": "min_chars: minimum chars before sending a block. idle_ms: max idle time before flushing.", + "group": "streaming", + }, + "ignore_attachments": { + "label": "Ignore Attachments", + "help_text": "Strip attachments from incoming messages before processing.", + "group": "processing", + }, + "ignore_stories": { + "label": "Ignore Stories", + "help_text": "Skip Signal story messages entirely.", + "group": "processing", + }, + "config_writes": { + "label": "Allow Config Writes", + "help_text": "Allow configuration changes via Signal messages. Disabled by default for security.", + "group": "security", + }, + "name": { + "label": "Channel Display Name", + "help_text": "Human-readable name for this Signal channel instance.", + "group": "general", + }, + "default_to": { + "label": "Default Send Target", + "help_text": "Default recipient when no explicit target is specified in send operations.", + "group": "general", + }, + "context_visibility_mode": { + "label": "Context Visibility", + "help_text": "Control message context sharing: 'all', 'same-group', 'trusted', or 'none'.", + "group": "security", + }, + "security": { + "label": "Security Policy", + "help_text": "DM policy (pairing/allowlist/open/disabled) and group policy (allowlist/open/disabled).", + "group": "security", + }, + "streaming": { + "label": "Streaming Config", + "help_text": "Streaming edit support: 'auto' (detect), 'disabled', or 'force'.", + "group": "streaming", + }, + "allow_remote_daemon": { + "label": "Allow Remote Daemon", + "help_text": "Allow connecting to non-loopback daemon URLs. Disables SSRF protection.", + "group": "daemon", + }, + "download_max_mb": { + "label": "Max Download Size (MB)", + "help_text": "Maximum signal-cli download size for attachments. Default: 256MB.", + "group": "messaging", + }, + "entity_cache_ttl": { + "label": "Entity Cache TTL (seconds)", + "help_text": "How long to cache user/group lookups. Default: 600s.", + "group": "processing", + }, + "event_queue_max_size": { + "label": "Event Queue Max Size", + "help_text": "Maximum number of events to queue before dropping oldest. Default: 1000.", + "group": "processing", + }, + "prefer_native_binary": { + "label": "Prefer Native Binary", + "help_text": "Download GraalVM native binary instead of universal tar.gz when auto-installing.", + "group": "daemon", + }, + "receive_mode_daemon_level": { + "label": "Daemon Receipt Mode", + "help_text": "How daemon handles read receipts: 'auto', 'on-start' (always use --send-read-receipts), or 'manual'.", + "group": "daemon", + }, + "human_delay": { + "label": "Human Typing Delay", + "help_text": "Simulate human typing: enabled (bool), min_ms (300), max_ms (1500). Adds randomized delay before sends.", + "group": "messaging", + }, + "ingest": { + "label": "Group Silent Ingest", + "help_text": "Log skipped group messages for analysis when enabled. Does not trigger agent replies.", + "group": "processing", + }, + "main_dm_owner_pin": { + "label": "Main DM Owner Pin", + "help_text": "Fix DM routing to a specific allowlisted sender (E.164 or UUID). Non-pinned DMs are silently skipped.", + "group": "security", + }, + "daemon_ready": { + "label": "Daemon Ready Timing", + "help_text": "poll_interval_ms (150), log_after_ms (10000), log_interval_ms (10000). Controls readiness check timing.", + "group": "daemon", + }, + "duplicate_reaction_check": { + "label": "Duplicate Reaction Check", + "help_text": "Deduplicate reactions using composite key (message, sender, emoji, group).", + "group": "reactions", + }, + "command_double_auth": { + "label": "Command Double-Auth", + "help_text": "Require authorization from both DM allowFrom AND group allowFrom for command execution.", + "group": "security", + }, + "pairing_store_persist": { + "label": "Pairing Store Persistence", + "help_text": "Persist pairing approvals for cross-restart survival. Requires store integration.", + "group": "security", + }, + "audit": { + "label": "Security Audit Logging", + "help_text": "Enable structured JSON audit logs for all message access decisions.", + "group": "security", + }, + "ai_vision": { + "label": "AI Media Vision", + "help_text": "Use vision-capable LLM to analyze image content and inject descriptions into message context.", + "group": "ai", + }, + "lane_separator": { + "label": "Reasoning Lane Separator", + "help_text": "Separator between main output and reasoning lane in streaming. Default: '---'.", + "group": "streaming", + }, +} diff --git a/backend/package/yuxi/channels/adapters/signal/daemon.py b/backend/package/yuxi/channels/adapters/signal/daemon.py new file mode 100644 index 00000000..5543c303 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/daemon.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +import asyncio +import logging +import shutil +from collections.abc import Callable, Awaitable + +from yuxi.channels.models import HealthStatus +from yuxi.channels.adapters.signal.client import RpcClient + +logger = logging.getLogger(__name__) + +DAEMON_STARTUP_RETRIES = 30 +DAEMON_STARTUP_RETRY_DELAY = 1.0 +DAEMON_STOP_TIMEOUT = 10.0 + + +class SignalDaemonManager: + def __init__( + self, + cli_path: str, + account: str, + http_listen: str = "127.0.0.1:8080", + home_dir: str | None = None, + java_opts: str | None = None, + receive_mode: str | None = None, + send_read_receipts: bool | None = None, + daemon_startup_retries: int = DAEMON_STARTUP_RETRIES, + auto_start: bool = True, + poll_interval_ms: int = 150, + log_after_ms: int = 10000, + log_interval_ms: int = 10000, + ): + self._cli_path = cli_path + self._account = account + self._http_listen = http_listen + self._home_dir = home_dir + self._java_opts = java_opts + self._receive_mode = receive_mode + self._send_read_receipts = send_read_receipts + self._daemon_startup_retries = daemon_startup_retries + self._auto_start = auto_start + self._poll_interval_ms = poll_interval_ms + self._log_after_ms = log_after_ms + self._log_interval_ms = log_interval_ms + self._process: asyncio.subprocess.Process | None = None + self._monitor_task: asyncio.Task | None = None + self._on_crash: asyncio.Event | None = None + self._crash_handler: Callable[[int, str], Awaitable[None]] | None = None + + @property + def listen_addr(self) -> str: + return self._http_listen + + @property + def is_running(self) -> bool: + return self._process is not None and self._process.returncode is None + + def on_crash(self, handler: Callable[[int, str], Awaitable[None]]) -> None: + self._crash_handler = handler + + async def start(self) -> None: + if not self._auto_start: + logger.info("signal-cli daemon auto_start is disabled, skipping") + return + + if not shutil.which(self._cli_path) and self._cli_path != "signal-cli": + raise FileNotFoundError(f"signal-cli not found at: {self._cli_path}") + + cmd = [self._cli_path, "-a", self._account, "daemon", "--http-listen", self._http_listen] + + if self._receive_mode: + cmd.extend(["--receive-mode", self._receive_mode]) + + if self._send_read_receipts is not None: + if self._send_read_receipts: + cmd.append("--send-read-receipts") + else: + cmd.append("--no-send-read-receipts") + + if self._home_dir: + cmd.extend(["--config", self._home_dir]) + + env = None + if self._java_opts: + env = {**dict(__import__("os").environ), "JAVA_OPTS": self._java_opts} + + logger.info(f"Starting signal-cli daemon: {' '.join(cmd)}") + + self._process = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=env, + ) + + self._on_crash = asyncio.Event() + + self._monitor_task = asyncio.create_task(self._monitor_process()) + + rpc_client = RpcClient(f"http://{self._http_listen}") + await rpc_client.connect() + + try: + poll_ms = max(100, self._poll_interval_ms) + total_elapsed_ms = 0 + timeout_ms = self._daemon_startup_retries * 1000 + logged = False + + while total_elapsed_ms < timeout_ms: + try: + await rpc_client.call("version") + logger.info("signal-cli daemon is ready") + return + except Exception: + await asyncio.sleep(poll_ms / 1000.0) + total_elapsed_ms += poll_ms + + if not logged and total_elapsed_ms >= self._log_after_ms: + logged = True + logger.info( + f"signal-cli daemon not ready after {total_elapsed_ms}ms, " + f"will log every {self._log_interval_ms}ms" + ) + elif logged and total_elapsed_ms % self._log_interval_ms < poll_ms: + logger.info(f"signal-cli daemon still not ready (elapsed={total_elapsed_ms}ms)") + + raise TimeoutError( + f"signal-cli daemon did not become ready within {timeout_ms}ms (poll_interval={poll_ms}ms)" + ) + finally: + await rpc_client.disconnect() + + async def stop(self) -> None: + if self._monitor_task: + self._monitor_task.cancel() + try: + await self._monitor_task + except asyncio.CancelledError: + pass + self._monitor_task = None + + if self._process: + logger.info("Stopping signal-cli daemon") + self._process.terminate() + try: + await asyncio.wait_for(self._process.wait(), timeout=DAEMON_STOP_TIMEOUT) + except TimeoutError: + logger.warning("signal-cli daemon did not stop gracefully, sending SIGKILL") + self._process.kill() + await self._process.wait() + self._process = None + + if self._on_crash: + self._on_crash.clear() + + async def restart(self) -> None: + logger.info("Restarting signal-cli daemon") + await self.stop() + await self.start() + + async def health_check(self) -> HealthStatus: + if not self.is_running: + return HealthStatus( + status="unhealthy", + last_error="signal-cli daemon process is not running", + ) + + try: + rpc_client = RpcClient(f"http://{self._http_listen}") + await rpc_client.connect() + try: + await rpc_client.call("version") + return HealthStatus(status="healthy") + finally: + await rpc_client.disconnect() + except Exception as e: + return HealthStatus(status="unhealthy", last_error=str(e)) + + async def _monitor_process(self) -> None: + while self._process and self._process.returncode is None: + try: + await asyncio.wait_for(self._process.wait(), timeout=5.0) + break + except TimeoutError: + continue + + if self._process: + exit_code = self._process.returncode + stderr_data = await self._process.stderr.read() if self._process.stderr else b"" + msg = stderr_data.decode("utf-8", errors="replace")[:500] + self._classify_and_log_stderr(msg) + logger.error(f"signal-cli daemon exited with code {exit_code}: {msg}") + if self._on_crash: + self._on_crash.set() + if self._crash_handler: + try: + await self._crash_handler(exit_code, msg) + except Exception: + logger.exception("Crash handler failed") + + @staticmethod + def _classify_and_log_stderr(stderr_text: str) -> None: + if not stderr_text: + return + lower = stderr_text.lower() + if any(kw in lower for kw in ("error", "exception", "fatal", "failed")): + logger.error(f"[signal-cli stderr] {stderr_text[:500]}") + elif any(kw in lower for kw in ("warn", "warning")): + logger.warning(f"[signal-cli stderr] {stderr_text[:500]}") + elif any(kw in lower for kw in ("debug", "trace")): + logger.debug(f"[signal-cli stderr] {stderr_text[:500]}") + else: + logger.info(f"[signal-cli stderr] {stderr_text[:500]}") diff --git a/backend/package/yuxi/channels/adapters/signal/directory.py b/backend/package/yuxi/channels/adapters/signal/directory.py new file mode 100644 index 00000000..383fe995 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/directory.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from yuxi.channels.adapters.signal.client import RpcClient, RpcError +from yuxi.channels.adapters.signal.identity import get_identities + + +async def list_peers( + rpc_client: RpcClient, + account: str, + query: str | None = None, + trust_level: str | None = None, +) -> list[dict]: + try: + identities = await get_identities(rpc_client, account) + result = [] + for i in identities: + entry = { + "number": i.get("number", ""), + "name": i.get("name", ""), + "trust_level": i.get("trustLevel", "UNTRUSTED"), + "fingerprint": i.get("fingerprint", ""), + } + if query and not _match_query(entry, query): + continue + if trust_level and entry["trust_level"].upper() != trust_level.upper(): + continue + result.append(entry) + return result + except RpcError: + return [] + + +async def list_groups( + rpc_client: RpcClient, + account: str, + name_filter: str | None = None, +) -> list[dict]: + try: + result = await rpc_client.call("listGroups", {"account": account}) + groups = result.get("groups", []) + entries = [] + for g in groups: + entry = { + "group_id": g.get("groupId", g.get("id", "")), + "name": g.get("name", ""), + "description": g.get("description", ""), + "member_count": g.get("memberCount", 0), + } + if name_filter and name_filter.lower() not in entry["name"].lower(): + continue + entries.append(entry) + return entries + except RpcError: + return [] + + +def _match_query(entry: dict, query: str) -> bool: + q = query.lower() + return ( + q in entry.get("number", "").lower() + or q in entry.get("name", "").lower() + or q in entry.get("fingerprint", "").lower() + ) diff --git a/backend/package/yuxi/channels/adapters/signal/entity_cache.py b/backend/package/yuxi/channels/adapters/signal/entity_cache.py new file mode 100644 index 00000000..705f3762 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/entity_cache.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +import asyncio +import time +from typing import Any + + +class EntityCache: + def __init__(self, ttl: int = 600, max_size: int = 500): + self._ttl = ttl + self._max_size = max_size + self._cache: dict[str, tuple[Any, float]] = {} + self._access_order: list[str] = [] + self._lock = asyncio.Lock() + + def _now(self) -> float: + return time.monotonic() + + async def get(self, key: str) -> Any | None: + async with self._lock: + entry = self._cache.get(key) + if entry is None: + return None + value, ts = entry + if self._now() - ts > self._ttl: + self._cache.pop(key, None) + if key in self._access_order: + self._access_order.remove(key) + return None + if key in self._access_order: + self._access_order.remove(key) + self._access_order.append(key) + return value + + async def set(self, key: str, value: Any) -> None: + async with self._lock: + if key in self._cache: + self._cache[key] = (value, self._now()) + if key in self._access_order: + self._access_order.remove(key) + self._access_order.append(key) + return + + if len(self._cache) >= self._max_size: + oldest = self._access_order.pop(0) + self._cache.pop(oldest, None) + + self._cache[key] = (value, self._now()) + self._access_order.append(key) + + async def delete(self, key: str) -> None: + async with self._lock: + self._cache.pop(key, None) + if key in self._access_order: + self._access_order.remove(key) + + async def clear(self) -> None: + async with self._lock: + self._cache.clear() + self._access_order.clear() + + async def stats(self) -> dict: + async with self._lock: + return { + "size": len(self._cache), + "max_size": self._max_size, + "ttl": self._ttl, + } diff --git a/backend/package/yuxi/channels/adapters/signal/event_queue.py b/backend/package/yuxi/channels/adapters/signal/event_queue.py new file mode 100644 index 00000000..9bd5263e --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/event_queue.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import asyncio +import heapq +import logging +from collections.abc import Awaitable, Callable + +logger = logging.getLogger(__name__) + + +class OrderedEventQueue: + def __init__(self, max_size: int = 1000, max_wait_ms: int = 5000): + self._max_size = max_size + self._max_wait_ms = max_wait_ms + self._heap: list[tuple[int, int, dict]] = [] + self._seq_counter = 0 + self._locked = False + self._consumer_task: asyncio.Task | None = None + self._handler: Callable[[dict], Awaitable[None]] | None = None + self._event = asyncio.Event() + + def set_handler(self, handler: Callable[[dict], Awaitable[None]]) -> None: + self._handler = handler + + async def push(self, event: dict, timestamp: int = 0) -> None: + self._seq_counter += 1 + entry = (timestamp, self._seq_counter, event) + heapq.heappush(self._heap, entry) + if len(self._heap) > self._max_size: + heapq.heappop(self._heap) + self._event.set() + + async def start(self) -> None: + if self._consumer_task is not None: + return + self._consumer_task = asyncio.create_task(self._consume_loop()) + + async def stop(self) -> None: + if self._consumer_task: + self._consumer_task.cancel() + try: + await self._consumer_task + except asyncio.CancelledError: + pass + self._consumer_task = None + + async def _consume_loop(self) -> None: + while True: + try: + if not self._heap: + await asyncio.wait_for( + self._event.wait(), + timeout=self._max_wait_ms / 1000.0, + ) + self._event.clear() + if not self._heap: + continue + + ts, seq, event = heapq.heappop(self._heap) + if self._handler: + try: + await self._handler(event) + except Exception: + logger.exception("OrderedEventQueue handler failed") + + except TimeoutError: + continue + except asyncio.CancelledError: + break + except Exception: + logger.exception("OrderedEventQueue consume loop error") + + @property + def size(self) -> int: + return len(self._heap) diff --git a/backend/package/yuxi/channels/adapters/signal/exec_auth.py b/backend/package/yuxi/channels/adapters/signal/exec_auth.py new file mode 100644 index 00000000..bcde9129 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/exec_auth.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from enum import StrEnum +from typing import Any +from collections.abc import Awaitable, Callable + + +class ExecAuthResult(StrEnum): + APPROVED = "approved" + DENIED = "denied" + PENDING = "pending" + + +class ExecAuthAdapter: + def __init__(self, auto_approve: bool = False): + self._auto_approve = auto_approve + self._pending_requests: dict[str, dict] = {} + self._request_counter = 0 + self._on_approval_request: Callable[[dict], Awaitable[None]] | None = None + + def on_approval_request(self, handler: Callable[[dict], Awaitable[None]]) -> None: + self._on_approval_request = handler + + async def request_approval(self, action: str, params: dict[str, Any], user_id: str) -> ExecAuthResult: + if self._auto_approve: + return ExecAuthResult.APPROVED + + self._request_counter += 1 + request_id = f"exec_auth_{self._request_counter}" + request = { + "id": request_id, + "action": action, + "params": params, + "user_id": user_id, + "result": ExecAuthResult.PENDING, + } + self._pending_requests[request_id] = request + + if self._on_approval_request: + await self._on_approval_request(request) + + return ExecAuthResult.PENDING + + def approve(self, request_id: str) -> ExecAuthResult: + request = self._pending_requests.get(request_id) + if not request: + return ExecAuthResult.DENIED + request["result"] = ExecAuthResult.APPROVED + return ExecAuthResult.APPROVED + + def deny(self, request_id: str) -> ExecAuthResult: + request = self._pending_requests.get(request_id) + if not request: + return ExecAuthResult.DENIED + request["result"] = ExecAuthResult.DENIED + return ExecAuthResult.DENIED + + def get_pending(self) -> list[dict]: + return [r for r in self._pending_requests.values() if r["result"] == ExecAuthResult.PENDING] diff --git a/backend/package/yuxi/channels/adapters/signal/format.py b/backend/package/yuxi/channels/adapters/signal/format.py new file mode 100644 index 00000000..8c3ecca0 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/format.py @@ -0,0 +1,292 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass, field + + +@dataclass +class StyleRange: + start: int + length: int + style: str + + +BOLD = "BOLD" +ITALIC = "ITALIC" +STRIKETHROUGH = "STRIKETHROUGH" +MONOSPACE = "MONOSPACE" +SPOILER = "SPOILER" + +_MARKDOWN_PATTERNS: list[tuple[str, str]] = [ + (r"\*\*(.+?)\*\*", BOLD), + (r"__(.+?)__", BOLD), + (r"\*(.+?)\*", ITALIC), + (r"_(.+?)_", ITALIC), + (r"~~(.+?)~~", STRIKETHROUGH), + (r"`(.+?)`", MONOSPACE), + (r"\|\|(.+?)\|\|", SPOILER), +] + +_HEADING_PATTERN = re.compile(r"^(#{1,6})\s+(.+)$", re.MULTILINE) +_BLOCKQUOTE_PATTERN = re.compile(r"^>\s?(.+)$", re.MULTILINE) +_TABLE_SEP_PATTERN = re.compile(r"^\|?[-:|\s]+\|?$") +_LINK_PATTERN = re.compile(r"\[([^\]]+)\]\(([^)]+)\)") +_URL_PATTERN = re.compile(r"https?://\S+") + + +@dataclass +class FormattedText: + body: str + styles: list[StyleRange] = field(default_factory=list) + + +def markdown_to_signal_styles( + text: str, + table_mode: str = "bullets", + heading_style: str = "bold", + blockquote_prefix: str = "> ", +) -> FormattedText: + result = text + styles: list[StyleRange] = [] + + result = _convert_tables(result, table_mode) + + result, link_ranges = _process_links(result) + + for pattern, style_name in _MARKDOWN_PATTERNS: + result, styles = _apply_markdown_pattern(result, pattern, style_name, styles, link_ranges) + + result = _convert_headings(result, heading_style) + result = _convert_blockquotes(result, blockquote_prefix) + + styles = _merge_adjacent_styles(sorted(styles, key=lambda s: s.start)) + + return FormattedText(body=result, styles=styles) + + +def _convert_tables(text: str, table_mode: str) -> str: + if table_mode != "bullets": + return text + + lines = text.split("\n") + result: list[str] = [] + in_table = False + table_rows: list[list[str]] = [] + + for line in lines: + stripped = line.strip() + if stripped.startswith("|") and stripped.endswith("|"): + if _TABLE_SEP_PATTERN.match(stripped): + continue + cells = [c.strip() for c in stripped[1:-1].split("|")] + table_rows.append(cells) + in_table = True + continue + else: + if in_table and table_rows: + result.extend(_table_rows_to_bullets(table_rows)) + table_rows = [] + in_table = False + result.append(line) + + if table_rows: + result.extend(_table_rows_to_bullets(table_rows)) + + return "\n".join(result) + + +def _table_rows_to_bullets(rows: list[list[str]]) -> list[str]: + if not rows: + return [] + bullets: list[str] = [] + is_header = True + for row in rows: + if is_header and len(rows) > 1: + bullets.append(" • ".join(row)) + is_header = False + else: + for cell in row: + bullets.append(f"• {cell}") + return bullets + + +def _process_links(text: str) -> tuple[str, list[tuple[int, int]]]: + link_replacements: list[tuple[int, int, str, str]] = [] + for m in _LINK_PATTERN.finditer(text): + label = m.group(1) + url = m.group(2) + start, end = m.span() + if label == url or _url_text_equivalent(label, url): + replacement = url + else: + replacement = f"{label} ({url})" + link_replacements.append((start, end, replacement, url)) + + result = text + offset = 0 + link_ranges: list[tuple[int, int]] = [] + for start, end, replacement, _url in sorted(link_replacements, key=lambda x: x[0]): + old_len = end - start + new_len = len(replacement) + actual_start = start + offset + result = result[:actual_start] + replacement + result[actual_start + old_len :] + link_ranges.append((actual_start, actual_start + new_len)) + offset += new_len - old_len + + return result, link_ranges + + +def _url_text_equivalent(label: str, url: str) -> bool: + clean_label = label.strip().rstrip("/").lower() + clean_url = url.strip().rstrip("/").lower() + if clean_label == clean_url: + return True + if clean_url.startswith("https://"): + if clean_label == clean_url.removeprefix("https://"): + return True + if clean_url.startswith("http://"): + if clean_label == clean_url.removeprefix("http://"): + return True + return False + + +def _apply_markdown_pattern( + text: str, + pattern: str, + style_name: str, + existing_styles: list[StyleRange], + link_ranges: list[tuple[int, int]], +) -> tuple[str, list[StyleRange]]: + compiled = re.compile(pattern) + replacements: list[tuple[int, int, str]] = [] + new_styles: list[StyleRange] = [] + + for m in compiled.finditer(text): + inner = m.group(1) + start, end = m.span() + if _overlaps_any(start, end, link_ranges): + continue + replacement = inner + replacements.append((start, end, replacement)) + new_styles.append(StyleRange(start=start, length=len(inner), style=style_name)) + + result = text + offset = 0 + final_styles = list(existing_styles) + + replacements.sort(key=lambda x: x[0]) + for start, end, replacement in replacements: + actual_start = start + offset + old_len = end - start + new_len = len(replacement) + result = result[:actual_start] + replacement + result[actual_start + old_len :] + + for ns in new_styles: + if ns.start == start: + ns.start = actual_start + offset += new_len - old_len + + for ns in new_styles: + if ns.start + ns.length <= len(result): + final_styles.append(ns) + + for existing in final_styles: + if existing in new_styles: + continue + for rep_start, rep_end, _ in replacements: + if existing.start >= rep_start and existing.start < rep_end: + existing.start += len(replacement) - (rep_end - rep_start) + + return result, final_styles + + +def _overlaps_any(start: int, end: int, ranges: list[tuple[int, int]]) -> bool: + for r_start, r_end in ranges: + if start < r_end and end > r_start: + return True + return False + + +def _convert_headings(text: str, heading_style: str) -> str: + if heading_style == "bold": + + def _replace(m: re.Match) -> str: + content = m.group(2) + return f"**{content}**" + + return _HEADING_PATTERN.sub(_replace, text) + return text + + +def _convert_blockquotes(text: str, blockquote_prefix: str) -> str: + lines = text.split("\n") + result = [] + in_blockquote = False + prefix = blockquote_prefix or "> " + + for line in lines: + m = _BLOCKQUOTE_PATTERN.match(line) + if m: + result.append(f"{prefix}{m.group(1)}") + in_blockquote = True + else: + if in_blockquote and line.strip() == "": + in_blockquote = False + result.append(line) + + return "\n".join(result) + + +def _merge_adjacent_styles(styles: list[StyleRange]) -> list[StyleRange]: + if not styles: + return [] + merged: list[StyleRange] = [] + for style in styles: + if merged and merged[-1].style == style.style and merged[-1].start + merged[-1].length == style.start: + merged[-1].length += style.length + else: + merged.append(StyleRange(start=style.start, length=style.length, style=style.style)) + return merged + + +def split_text(text: str, limit: int = 4000, chunk_mode: str = "newline") -> list[str]: + if len(text) <= limit: + return [text] + + if chunk_mode == "length": + return [text[i : i + limit] for i in range(0, len(text), limit)] + + chunks: list[str] = [] + paragraphs = text.split("\n\n") + current = "" + + for para in paragraphs: + if len(current) + len(para) + 2 <= limit: + current = f"{current}\n\n{para}" if current else para + else: + if current: + chunks.append(current) + if len(para) > limit: + for i in range(0, len(para), limit): + chunks.append(para[i : i + limit]) + current = "" + else: + current = para + + if current: + chunks.append(current) + return chunks or [text] + + +def clamp_styles_to_length(styles: list[StyleRange], body: str) -> list[StyleRange]: + body_len = len(body) + result: list[StyleRange] = [] + for s in styles: + if s.start >= body_len: + continue + end = s.start + s.length + if end > body_len: + s.length = body_len - s.start + if s.length > 0: + result.append(s) + return result diff --git a/backend/package/yuxi/channels/adapters/signal/identity.py b/backend/package/yuxi/channels/adapters/signal/identity.py new file mode 100644 index 00000000..2ae368d0 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/identity.py @@ -0,0 +1,41 @@ +import logging + +from yuxi.channels.adapters.signal.client import RpcClient, RpcError + +logger = logging.getLogger(__name__) + + +async def get_identities(rpc_client: RpcClient, account: str) -> list[dict]: + try: + result = await rpc_client.call("getIdentities", {"account": account}) + return result.get("identities", []) + except RpcError as e: + logger.warning(f"Failed to get identities: {e}") + return [] + + +async def trust_identity( + rpc_client: RpcClient, + account: str, + recipient: str, + fingerprint: str | None = None, +) -> bool: + try: + params: dict = {"account": account, "recipient": recipient} + if fingerprint: + params["fingerprint"] = fingerprint + else: + params["trustAllKeys"] = True + await rpc_client.call("trustIdentity", params) + return True + except RpcError as e: + logger.warning(f"Failed to trust identity {recipient}: {e}") + return False + + +async def get_self_number(rpc_client: RpcClient, account: str) -> str | None: + try: + result = await rpc_client.call("getSelf", {"account": account}) + return result.get("number") + except RpcError: + return account diff --git a/backend/package/yuxi/channels/adapters/signal/install.py b/backend/package/yuxi/channels/adapters/signal/install.py new file mode 100644 index 00000000..63d7a56f --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/install.py @@ -0,0 +1,210 @@ +import logging +import os +import platform +import shutil +import subprocess +import tempfile +from urllib.request import urlopen, Request +import json + +logger = logging.getLogger(__name__) + + +INSTALL_GUIDE_URL = "https://github.com/AsamK/signal-cli/releases" +GITHUB_API_RELEASES = "https://api.github.com/repos/AsamK/signal-cli/releases/latest" +DEFAULT_MAX_DOWNLOAD_MB = 256 + + +def _get_platform_suffix() -> str | None: + system = platform.system().lower() + machine = platform.machine().lower() + if system == "linux" and machine in ("x86_64", "amd64"): + return "linux-x86_64" + if system == "linux" and machine in ("aarch64", "arm64"): + return "linux-aarch64" + return None + + +async def fetch_latest_release_url(prefer_native: bool = True) -> str | None: + try: + req = Request(GITHUB_API_RELEASES, headers={"User-Agent": "ForcePilot"}) + with urlopen(req, timeout=15) as resp: + data = json.loads(resp.read().decode()) + assets = data.get("assets", []) + native_suffix = _get_platform_suffix() if prefer_native else None + + native_candidate = None + universal_candidate = None + for asset in assets: + name = asset.get("name", "") + if native_suffix and native_suffix in name.lower() and name.endswith(".tar.gz") and "signal-cli" in name: + native_candidate = asset.get("browser_download_url") + if name.endswith(".tar.gz") and "signal-cli" in name: + universal_candidate = asset.get("browser_download_url") + + return native_candidate or universal_candidate + except Exception as e: + logger.warning(f"Failed to fetch latest signal-cli release: {e}") + return None + + +async def auto_install_signal_cli( + target_dir: str | None = None, max_download_mb: int = DEFAULT_MAX_DOWNLOAD_MB +) -> bool: + url = await fetch_latest_release_url() + if not url: + logger.error("Could not find signal-cli download URL") + return False + + if target_dir is None: + target_dir = os.path.join(os.path.expanduser("~"), ".local", "signal-cli") + + os.makedirs(target_dir, exist_ok=True) + + logger.info(f"Downloading signal-cli from {url}") + + try: + req = Request(url, headers={"User-Agent": "ForcePilot"}) + with urlopen(req, timeout=300) as resp: + content_length = resp.headers.get("Content-Length") + if content_length: + size_mb = int(content_length) / (1024 * 1024) + if size_mb > max_download_mb: + max_size_bytes = max_download_mb * 1024 * 1024 + raise ValueError( + f"signal-cli download size ({size_mb:.1f} MB) exceeds limit ({max_download_mb} MB)." + ) + data = resp.read() + if len(data) > max_download_mb * 1024 * 1024: + raise ValueError( + f"signal-cli download size ({len(data) / (1024 * 1024):.1f} MB) exceeds limit ({max_download_mb} MB)." + ) + except Exception as e: + logger.error(f"Failed to download signal-cli: {e}") + return False + + with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as tmp: + tmp.write(data) + tarball = tmp.name + + try: + import tarfile + + with tarfile.open(tarball, "r:gz") as tar: + tar.extractall(target_dir) + logger.info(f"signal-cli installed to {target_dir}") + return True + except Exception as e: + logger.error(f"Failed to extract signal-cli: {e}") + return False + finally: + try: + os.unlink(tarball) + except OSError: + pass + + +def check_signal_cli_installed(cli_path: str = "signal-cli") -> tuple[bool, str | None]: + if shutil.which(cli_path): + try: + result = subprocess.run( + [cli_path, "--version"], + capture_output=True, + text=True, + timeout=10, + ) + if result.returncode == 0: + version = result.stdout.strip() or result.stderr.strip() + logger.info(f"signal-cli found: {version}") + return True, version + except (subprocess.TimeoutExpired, FileNotFoundError, OSError): + pass + + return False, None + + +def check_java_installed() -> tuple[bool, str | None]: + java_home = shutil.which("java") + if not java_home: + return False, None + + try: + result = subprocess.run( + [java_home, "-version"], + capture_output=True, + text=True, + timeout=10, + ) + version_output = result.stderr or result.stdout + logger.info(f"Java found: {version_output.strip().split(chr(10))[0]}") + return True, version_output.strip() + except (subprocess.TimeoutExpired, FileNotFoundError, OSError): + return False, None + + +def get_install_instructions() -> str: + system = platform.system().lower() + + if system == "linux": + return ( + "Install signal-cli on Linux:\n" + " 1. Download the latest release from:\n" + f" {INSTALL_GUIDE_URL}\n" + " 2. Install the .deb package:\n" + " sudo dpkg -i signal-cli_*.deb\n" + " Or use the universal tar:\n" + " tar xf signal-cli-*.tar.gz -C /usr/local\n" + " ln -s /usr/local/signal-cli-*/bin/signal-cli /usr/local/bin/signal-cli\n" + "\n" + "Requires Java 17+ JRE:\n" + " sudo apt install openjdk-17-jre-headless" + ) + elif system == "darwin": + return ( + "Install signal-cli on macOS:\n" + " brew install signal-cli\n" + "\n" + "Requires Java 17+ JRE:\n" + " brew install openjdk@17" + ) + else: + return ( + "Install signal-cli on Windows:\n" + f" 1. Download the latest release from {INSTALL_GUIDE_URL}\n" + " 2. Extract signal-cli-*.tar.gz\n" + " 3. Run signal-cli.bat from the bin directory\n" + "\n" + "Requires Java 17+ JRE:\n" + " Download from https://adoptium.net/" + ) + + +def print_install_guide() -> None: + java_ok, java_version = check_java_installed() + cli_ok, cli_version = check_signal_cli_installed() + + print("=== Signal CLI Environment Check ===") + print(f" Java 17+: {'OK' if java_ok else 'NOT FOUND'}") + if java_version: + print(f" {java_version.strip().split(chr(10))[0]}") + print(f" signal-cli: {'OK' if cli_ok else 'NOT FOUND'}") + if cli_version: + print(f" version: {cli_version}") + print() + + if not java_ok: + print("[!] Java 17+ JRE is required to run signal-cli.") + print() + if not cli_ok: + print("[!] signal-cli is not installed.") + print(get_install_instructions()) + else: + print("signal-cli is ready to use.") + print() + print("Next steps:") + print(" 1. Register your phone number:") + print(" signal-cli -a +1234567890 register") + print(" 2. Verify with the code received via SMS:") + print(" signal-cli -a +1234567890 verify ") + print(" 3. Start daemon mode:") + print(" signal-cli -a +1234567890 daemon --http-listen 127.0.0.1:8080") diff --git a/backend/package/yuxi/channels/adapters/signal/media_vision.py b/backend/package/yuxi/channels/adapters/signal/media_vision.py new file mode 100644 index 00000000..6d283e57 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/media_vision.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import base64 +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +VISION_SYSTEM_PROMPT = ( + "You are a media analyst. Describe the content of the provided media concisely. " + "For images: describe what you see. " + "For videos: note it's a video and describe visible elements. " + "Keep responses under 200 characters." +) + + +class MediaVisionAnalyzer: + def __init__(self, llm_call_fn: Any, enabled: bool = True): + self._llm_call = llm_call_fn + self._enabled = enabled + + async def analyze_image(self, image_data: bytes, mime_type: str = "image/jpeg") -> str | None: + if not self._enabled or not self._llm_call: + return None + + try: + b64 = base64.b64encode(image_data).decode("utf-8") + result = await self._llm_call( + messages=[ + {"role": "system", "content": VISION_SYSTEM_PROMPT}, + { + "role": "user", + "content": [ + {"type": "text", "text": "Describe this media content briefly."}, + {"type": "image_url", "image_url": {"url": f"data:{mime_type};base64,{b64}"}}, + ], + }, + ] + ) + return result.get("content", "") if isinstance(result, dict) else str(result) + except Exception: + logger.exception("Media vision analysis failed") + return None + + async def analyze_attachment( + self, + data: bytes, + filename: str | None = None, + mime_type: str = "application/octet-stream", + ) -> str | None: + if mime_type.startswith("image/"): + return await self.analyze_image(data, mime_type) + if mime_type.startswith("video/"): + return f"[video: {filename or 'unnamed'}]" + if mime_type.startswith("audio/"): + return f"[audio: {filename or 'unnamed'}]" + return f"[file: {filename or 'unnamed'}]" diff --git a/backend/package/yuxi/channels/adapters/signal/message_actions.py b/backend/package/yuxi/channels/adapters/signal/message_actions.py new file mode 100644 index 00000000..8c432519 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/message_actions.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +from typing import Any + + +SIGNAL_MESSAGE_ACTIONS = { + "send": { + "description": "Send a text message to a Signal chat", + "params": { + "recipient": {"type": "str", "description": "Recipient E.164 number or group:ID"}, + "message_body": {"type": "str", "description": "Message content"}, + "reply_to_id": {"type": "str", "description": "Message ID to reply to"}, + }, + }, + "react": { + "description": "React to a Signal message with an emoji", + "params": { + "recipient": {"type": "str", "description": "Recipient E.164 number or group:ID"}, + "target_sent_timestamp": {"type": "int", "description": "Timestamp of target message"}, + "emoji": {"type": "str", "description": "Emoji reaction"}, + "remove": {"type": "bool", "description": "Whether to remove the reaction"}, + }, + }, + "edit": { + "description": "Edit a previously sent message", + "params": { + "recipient": {"type": "str", "description": "Recipient E.164 number or group:ID"}, + "target_sent_timestamp": {"type": "int", "description": "Timestamp of target message"}, + "new_body": {"type": "str", "description": "New message content"}, + }, + }, + "delete": { + "description": "Delete a previously sent message", + "params": { + "recipient": {"type": "str", "description": "Recipient E.164 number or group:ID"}, + "timestamps": {"type": "list[int]", "description": "Timestamps of messages to delete"}, + }, + }, + "send_media": { + "description": "Send a media file (image/video/audio/file) to a Signal chat", + "params": { + "recipient": {"type": "str", "description": "Recipient E.164 number or group:ID"}, + "media_data": {"type": "bytes", "description": "Base64-encoded media data"}, + "media_type": {"type": "str", "description": "Media type: image/video/audio/file"}, + "filename": {"type": "str", "description": "Filename"}, + "caption": {"type": "str", "description": "Caption text"}, + }, + }, + "send_typing": { + "description": "Send typing indicator to a Signal chat", + "params": { + "recipient": {"type": "str", "description": "Recipient E.164 number or group:ID"}, + }, + }, + "send_read_receipt": { + "description": "Send read receipt for messages", + "params": { + "recipient": {"type": "str", "description": "Recipient E.164 number or group:ID"}, + "timestamps": {"type": "list[int]", "description": "Timestamps of messages to mark as read"}, + }, + }, + "get_user_info": { + "description": "Get identity info for a Signal user", + "params": { + "channel_user_id": {"type": "str", "description": "E.164 number of the user"}, + }, + }, +} + + +def describe_message_tools(actions_config: dict[str, Any] | None = None) -> list[dict[str, Any]]: + enabled = actions_config or {} + tools = [] + for action_name, action_def in SIGNAL_MESSAGE_ACTIONS.items(): + if enabled.get(action_name, True): + tools.append( + { + "name": f"signal_{action_name}", + "description": action_def["description"], + "parameters": action_def["params"], + } + ) + return tools diff --git a/backend/package/yuxi/channels/adapters/signal/monitor.py b/backend/package/yuxi/channels/adapters/signal/monitor.py new file mode 100644 index 00000000..c465637b --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/monitor.py @@ -0,0 +1,241 @@ +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import Callable, Awaitable + +from yuxi.channels.models import ChannelMessage, EventType +from yuxi.channels.adapters.signal.client import RpcClient +from yuxi.channels.adapters.signal.event_queue import OrderedEventQueue +from yuxi.channels.adapters.signal.normalize import ( + build_dedup_key, + check_and_add_dedup, + check_debounce, + is_own_message, + is_sync_message, + parse_signal_delete, + parse_signal_message, + parse_signal_reaction, +) +from yuxi.channels.adapters.signal.sse_reconnect import sse_event_stream + +logger = logging.getLogger(__name__) + + +class SSEMonitor: + def __init__( + self, + rpc_client: RpcClient, + account_number: str, + account_uuid: str | None = None, + ignore_attachments: bool = False, + ignore_stories: bool = False, + debounce_interval_ms: int = 0, + duplicate_reaction_check: bool = True, + event_queue: OrderedEventQueue | None = None, + sent_message_cache: dict[str, dict] | None = None, + ): + self._rpc = rpc_client + self._account = account_number + self._account_uuid = account_uuid + self._ignore_attachments = ignore_attachments + self._ignore_stories = ignore_stories + self._debounce_interval_ms = debounce_interval_ms + self._duplicate_reaction_check = duplicate_reaction_check + self._task: asyncio.Task | None = None + self._message_handler: Callable[[ChannelMessage], Awaitable[None]] | None = None + self._running = False + self._reaction_seen: set[str] = set() + self._event_queue = event_queue + self._sent_message_cache = sent_message_cache + + if self._event_queue is not None: + self._event_queue.set_handler(self._dispatch_from_queue) + + def on_message(self, handler: Callable[[ChannelMessage], Awaitable[None]]) -> None: + self._message_handler = handler + + async def start(self) -> None: + self._running = True + if self._event_queue is not None: + await self._event_queue.start() + url = f"{self._rpc.base_url}/api/v1/events" + self._task = asyncio.create_task(sse_event_stream(url, {"account": self._account}, self._handle_sse_event)) + + async def stop(self) -> None: + self._running = False + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + if self._event_queue is not None: + await self._event_queue.stop() + + async def _dispatch_from_queue(self, event: dict) -> None: + msg = self._parse_event(event) + if msg and self._message_handler: + await self._message_handler(msg) + + async def _handle_sse_event(self, event_data: str) -> None: + try: + data = json.loads(event_data) + except json.JSONDecodeError: + logger.warning(f"Failed to parse SSE event: {event_data[:200]}") + return + + if is_sync_message(data): + return + + if self._ignore_stories and self._is_story_event(data): + return + + if is_own_message(data, self._account, self._account_uuid): + return + + self._check_sent_message_cache(data) + + dedup_key = build_dedup_key(data, self._account) + if dedup_key and check_and_add_dedup(dedup_key): + return + + conversation = data.get("dataMessage", {}).get("groupInfo", {}).get("groupId") or data.get("envelope", {}).get( + "source", "" + ) + if conversation and check_debounce(conversation, self._debounce_interval_ms): + return + + if self._ignore_attachments: + data = self._strip_attachments(data) + + if self._is_receipt_event(data): + self._log_receipt(data) + return + + if self._event_queue is not None: + ts = self._extract_timestamp(data) + await self._event_queue.push(data, timestamp=ts) + return + + msg = self._parse_event(data) + if msg and self._message_handler: + await self._message_handler(msg) + + def _parse_event(self, data: dict) -> ChannelMessage | None: + msg: ChannelMessage | None = None + + if self._is_message_event(data): + msg = parse_signal_message(data) + elif self._is_edit_message_event(data): + msg = parse_signal_message(self._extract_edit_data(data)) + if msg: + msg.event_type = EventType.MESSAGE_UPDATED + elif self._is_reaction_event(data): + if self._duplicate_reaction_check: + dedup_key = self._build_reaction_dedup_key(data) + if dedup_key and dedup_key in self._reaction_seen: + return None + if dedup_key: + self._reaction_seen.add(dedup_key) + msg = parse_signal_reaction(data) + elif self._is_delete_event(data): + msg = parse_signal_delete(data) + + return msg + + def _check_sent_message_cache(self, data: dict) -> None: + if not self._sent_message_cache: + return + dm = data.get("dataMessage", {}) + ts = dm.get("timestamp") + if not ts: + return + for msg_id, entry in self._sent_message_cache.items(): + if str(ts) in msg_id: + sender = data.get("envelope", {}).get("source", "unknown") + logger.info( + f"[Signal] Sent message delivery confirmed: ts={ts}, " + f"recipient={entry.get('recipient', 'unknown')}, sender={sender}" + ) + break + + @staticmethod + def _extract_timestamp(data: dict) -> int: + dm = data.get("dataMessage", {}) + ts = dm.get("timestamp", 0) + if ts: + return ts + reaction = data.get("reaction", {}) + ts = reaction.get("targetSentTimestamp", 0) + if ts: + return ts + delete_msg = data.get("deleteMessage", {}) + ts = delete_msg.get("targetSentTimestamp", 0) + return ts + + def _log_receipt(self, data: dict) -> None: + receipt = data.get("receiptMessage", {}) + receipt_type = receipt.get("type", "UNKNOWN") + timestamps = receipt.get("timestamps", []) + source = data.get("envelope", {}).get("source", "unknown") + logger.debug( + f"Signal receipt: type={receipt_type}, from={source}, " + f"timestamps_count={len(timestamps)}, " + f"timestamp_range={min(timestamps) if timestamps else 'N/A'}-{max(timestamps) if timestamps else 'N/A'}" + ) + + @staticmethod + def _build_reaction_dedup_key(data: dict) -> str | None: + envelope = data.get("envelope", {}) + reaction = data.get("reaction", {}) + message_id = str(reaction.get("targetSentTimestamp", "")) + sender_id = envelope.get("source", "") + emoji = reaction.get("emoji", "") + group_info = reaction.get("groupInfo", {}) or data.get("dataMessage", {}).get("groupInfo", {}) or {} + group_id = group_info.get("groupId", "") if isinstance(group_info, dict) else "" + if not message_id or not sender_id: + return None + return f"reaction:{message_id}:{sender_id}:{emoji}:{group_id}" + + @staticmethod + def _is_message_event(data: dict) -> bool: + return "envelope" in data and "dataMessage" in data + + @staticmethod + def _is_edit_message_event(data: dict) -> bool: + envelope = data.get("envelope", {}) + return "editMessage" in envelope and "dataMessage" in envelope.get("editMessage", {}) + + @staticmethod + def _extract_edit_data(data: dict) -> dict: + edit_msg = data.get("envelope", {}).get("editMessage", {}) + return { + "envelope": data.get("envelope", {}), + "dataMessage": edit_msg.get("dataMessage", {}), + } + + @staticmethod + def _is_reaction_event(data: dict) -> bool: + return "envelope" in data and "reaction" in data + + @staticmethod + def _is_delete_event(data: dict) -> bool: + return "envelope" in data and "deleteMessage" in data + + @staticmethod + def _is_receipt_event(data: dict) -> bool: + return "envelope" in data and "receiptMessage" in data + + @staticmethod + def _is_story_event(data: dict) -> bool: + return "envelope" in data and "storyMessage" in data + + @staticmethod + def _strip_attachments(data: dict) -> dict: + data_msg = data.get("dataMessage") + if isinstance(data_msg, dict) and "attachments" in data_msg: + data_msg = {**data_msg, "attachments": []} + data = {**data, "dataMessage": data_msg} + return data diff --git a/backend/package/yuxi/channels/adapters/signal/normalize.py b/backend/package/yuxi/channels/adapters/signal/normalize.py new file mode 100644 index 00000000..38bd07b1 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/normalize.py @@ -0,0 +1,376 @@ +from __future__ import annotations + +import re +import time +from datetime import datetime, UTC + +from yuxi.channels.models import ( + Attachment, + ChannelIdentity, + ChannelMessage, + ChannelType, + ChatType, + EventType, + MentionsInfo, + MessageType, +) + +E164_PATTERN = re.compile(r"^\+\d{7,15}$") +UUID_PREFIX = "uuid:" +GROUP_PREFIX = "group:" + +DEDUP_TTL = 60 +_dedup_store: dict[str, float] = {} + + +def is_own_message(data: dict, account_number: str, account_uuid: str | None = None) -> bool: + envelope = data.get("envelope", {}) + source = envelope.get("source", "") + if normalize_target(source) == normalize_target(account_number): + return True + if account_uuid: + source_uuid = envelope.get("sourceUuid", "") + if source_uuid and source_uuid == account_uuid: + return True + return False + + +def is_sync_message(data: dict) -> bool: + return "syncMessage" in data + + +DEBOUNCE_TTL = 5 +_debounce_store: dict[str, float] = {} + + +def check_debounce(conversation_key: str, interval_ms: int = 0) -> bool: + if interval_ms <= 0: + return False + now = time.monotonic() + _prune_debounce_store(now) + if conversation_key in _debounce_store: + last = _debounce_store[conversation_key] + if (now - last) * 1000 < interval_ms: + return True + _debounce_store[conversation_key] = now + return False + + +def _prune_debounce_store(now: float) -> None: + expired = [k for k, ts in _debounce_store.items() if now - ts > DEBOUNCE_TTL] + for k in expired: + _debounce_store.pop(k, None) + + +def build_dedup_key(data: dict, account_id: str) -> str | None: + envelope = data.get("envelope", {}) + source = envelope.get("source", "") + timestamp = ( + data.get("dataMessage", {}).get("timestamp") + or data.get("reaction", {}).get("targetSentTimestamp") + or data.get("deleteMessage", {}).get("targetSentTimestamp") + or data.get("receiptMessage", {}).get("timestamps", [None])[0] + ) + if not source or not timestamp: + return None + group_id = data.get("dataMessage", {}).get("groupInfo", {}).get("groupId") or "" + conversation = group_id or source + return f"signal:{account_id}:{conversation}:{source}:{timestamp}" + + +def check_and_add_dedup(key: str) -> bool: + now = time.monotonic() + _prune_dedup_store(now) + if key in _dedup_store: + return True + _dedup_store[key] = now + return False + + +def _prune_dedup_store(now: float) -> None: + expired = [k for k, ts in _dedup_store.items() if now - ts > DEDUP_TTL] + for k in expired: + _dedup_store.pop(k, None) + + +def normalize_target(raw: str) -> str: + stripped = raw.strip() + + if stripped.startswith(GROUP_PREFIX): + return stripped + if stripped.startswith(UUID_PREFIX): + return stripped + if E164_PATTERN.match(stripped): + return stripped + + digits = re.sub(r"[^\d]", "", stripped) + if 7 <= len(digits) <= 15: + return f"+{digits}" + + raise ValueError(f"Unable to normalize Signal target: {raw}") + + +def normalize_e164(raw: str) -> str: + stripped = raw.strip().removeprefix("+") + return re.sub(r"[^\d]", "", stripped) + + +def parse_signal_message(data: dict, channel_id: str = "signal") -> ChannelMessage | None: + envelope = data.get("envelope", {}) + data_msg = data.get("dataMessage", {}) + + if not data_msg: + return None + + source = envelope.get("source", "unknown") + source_name = envelope.get("sourceName", "") + group_info = data_msg.get("groupInfo", {}) + group_id = group_info.get("groupId") + chat_id = f"group:{group_id}" if group_id else source + timestamp_raw = data_msg.get("timestamp", 0) + ts = datetime.fromtimestamp(timestamp_raw / 1000.0, tz=UTC) if timestamp_raw else datetime.now(tz=UTC) + + content = data_msg.get("message", "") or data_msg.get("body", "") + chat_type = ChatType.GROUP if group_id else ChatType.DIRECT + + message_type = MessageType.TEXT + attachments: list[Attachment] = [] + if data_msg.get("attachments"): + raw_attachments = data_msg["attachments"] if isinstance(data_msg["attachments"], list) else [] + for att in raw_attachments: + content_type = att.get("contentType", "") + attachments.append( + Attachment( + type=_attachment_type(content_type), + file_id=att.get("id", ""), + filename=att.get("filename"), + mime_type=content_type, + size_bytes=att.get("size"), + metadata={"digest": att.get("digest", "")}, + ) + ) + if attachments: + message_type = _attachment_message_type(attachments[0].mime_type or "") + + reply_to_id = None + quote = data_msg.get("quote") + if quote: + reply_to_id = str(quote.get("id", "")) + + mentions = _extract_mentions(data_msg, content) + extracted_urls = _extract_urls(data_msg, content) + + metadata: dict = {} + if data_msg.get("expiresInSeconds"): + metadata["expires_in_seconds"] = data_msg["expiresInSeconds"] + if data_msg.get("viewOnce"): + metadata["view_once"] = True + if data_msg.get("sticker"): + metadata["sticker_pack_id"] = data_msg["sticker"].get("packId", "") + metadata["sticker_id"] = data_msg["sticker"].get("stickerId", "") + if data_msg.get("forwarded"): + metadata["forwarded"] = True + if data_msg.get("contacts"): + contacts = data_msg["contacts"] + metadata["contacts"] = contacts if isinstance(contacts, list) else [contacts] + if data_msg.get("location"): + loc = data_msg["location"] + metadata["location"] = { + "latitude": loc.get("latitude"), + "longitude": loc.get("longitude"), + "label": loc.get("label", ""), + } + if source_name: + metadata["sender_display_name"] = source_name + + if data_msg.get("sticker"): + message_type = MessageType.STICKER + elif data_msg.get("location"): + message_type = MessageType.LOCATION + elif attachments: + message_type = _attachment_message_type(attachments[0].mime_type or "") + + return ChannelMessage( + identity=ChannelIdentity( + channel_id=channel_id, + channel_type=ChannelType.SIGNAL, + channel_user_id=source, + channel_chat_id=chat_id, + channel_message_id=str(timestamp_raw), + ), + event_type=EventType.MESSAGE_RECEIVED, + message_type=message_type, + chat_type=chat_type, + content=content, + reply_to_message_id=reply_to_id, + attachments=attachments, + mentions=mentions, + extracted_urls=extracted_urls, + metadata=metadata, + timestamp=ts, + ) + + +def parse_signal_reaction(data: dict, channel_id: str = "signal") -> ChannelMessage | None: + envelope = data.get("envelope", {}) + reaction = data.get("reaction", {}) + + if not reaction: + return None + + source = envelope.get("source", "unknown") + target_author = reaction.get("targetAuthor", source) + emoji = reaction.get("emoji", "") + is_remove = reaction.get("remove", False) + target_timestamp = reaction.get("targetSentTimestamp", 0) + group_info = reaction.get("groupInfo", {}) + group_id = group_info.get("groupId") + chat_id = f"group:{group_id}" if group_id else source + + return ChannelMessage( + identity=ChannelIdentity( + channel_id=channel_id, + channel_type=ChannelType.SIGNAL, + channel_user_id=source, + channel_chat_id=chat_id, + channel_message_id=str(target_timestamp), + ), + event_type=EventType.MESSAGE_UPDATED, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP if group_id else ChatType.DIRECT, + content="(reaction_removed)" if is_remove else f"(reacted: {emoji})", + metadata={ + "reaction_emoji": emoji, + "reaction_removed": is_remove, + "reaction_target_author": target_author, + }, + ) + + +def parse_signal_receipt(data: dict, channel_id: str = "signal") -> ChannelMessage | None: + envelope = data.get("envelope", {}) + receipt = data.get("receiptMessage", {}) + + if not receipt: + return None + + source = envelope.get("source", "unknown") + receipt_type = receipt.get("type", "UNKNOWN") + timestamps = receipt.get("timestamps", []) + + return ChannelMessage( + identity=ChannelIdentity( + channel_id=channel_id, + channel_type=ChannelType.SIGNAL, + channel_user_id=source, + channel_chat_id=source, + ), + event_type=EventType.READ_RECEIPT, + message_type=MessageType.TEXT, + chat_type=ChatType.DIRECT, + content=f"(receipt: {receipt_type})", + metadata={ + "receipt_type": receipt_type, + "receipt_timestamps": timestamps, + }, + ) + + +def _attachment_type(content_type: str) -> str: + if content_type.startswith("image/"): + return "image" + if content_type.startswith("video/"): + return "video" + if content_type.startswith("audio/"): + return "audio" + return "file" + + +def _attachment_message_type(mime_type: str) -> MessageType: + if mime_type.startswith("image/"): + return MessageType.IMAGE + if mime_type.startswith("video/"): + return MessageType.VIDEO + if mime_type.startswith("audio/"): + return MessageType.AUDIO + return MessageType.FILE + + +URL_PATTERN = re.compile(r"https?://\S+") + + +def _extract_mentions(data_msg: dict, content: str, account_uuid: str | None = None) -> MentionsInfo | None: + body_ranges = data_msg.get("bodyRanges") + if not body_ranges or not isinstance(body_ranges, list): + return None + + mentioned_ids: list[str] = [] + is_bot_mentioned = False + for br in body_ranges: + mention_uuid = br.get("mentionUuid") + if mention_uuid: + mentioned_ids.append(mention_uuid) + if account_uuid and mention_uuid == account_uuid: + is_bot_mentioned = True + elif br.get("startsWith") and content: + start = br.get("start", 0) + length = br.get("length", 0) + if 0 <= start < len(content): + mention_text = content[start : start + length] + mentioned_ids.append(mention_text) + + if not mentioned_ids: + return None + + return MentionsInfo( + mentioned_user_ids=mentioned_ids, + is_bot_mentioned=is_bot_mentioned, + raw_text=content, + ) + + +def _extract_urls(data_msg: dict, content: str) -> list[str]: + previews = data_msg.get("previews") + if previews and isinstance(previews, list): + return [p.get("url", "") for p in previews if p.get("url")] + + return URL_PATTERN.findall(content) if content else [] + + +def parse_signal_delete(data: dict, channel_id: str = "signal") -> ChannelMessage | None: + envelope = data.get("envelope", {}) + delete_msg = data.get("deleteMessage") + + if not delete_msg: + return None + + source = envelope.get("source", "unknown") + target_timestamp = delete_msg.get("targetSentTimestamp", 0) + group_info = delete_msg.get("groupInfo", {}) + group_id = group_info.get("groupId") + chat_id = f"group:{group_id}" if group_id else source + + return ChannelMessage( + identity=ChannelIdentity( + channel_id=channel_id, + channel_type=ChannelType.SIGNAL, + channel_user_id=source, + channel_chat_id=chat_id, + channel_message_id=str(target_timestamp), + ), + event_type=EventType.MESSAGE_DELETED, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP if group_id else ChatType.DIRECT, + content="(message deleted)", + metadata={"deleted_timestamp": target_timestamp}, + ) + + +def looks_like_signal_target_id(raw: str) -> bool: + stripped = raw.strip() + if stripped.startswith(GROUP_PREFIX) or stripped.startswith(UUID_PREFIX): + return True + if E164_PATTERN.match(stripped): + return True + digits = re.sub(r"[^\d]", "", stripped) + return 7 <= len(digits) <= 15 diff --git a/backend/package/yuxi/channels/adapters/signal/outbound_session.py b/backend/package/yuxi/channels/adapters/signal/outbound_session.py new file mode 100644 index 00000000..570f58bb --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/outbound_session.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from yuxi.channels.models import ChatType +from yuxi.channels.adapters.signal.normalize import normalize_target + + +@dataclass +class OutboundSession: + peer: str + chat_type: ChatType + from_: str + to: str + + @classmethod + def resolve(cls, target: str, account_number: str) -> OutboundSession: + normalized = normalize_target(target) + if normalized.startswith("group:"): + return cls( + peer=normalized, + chat_type=ChatType.GROUP, + from_=account_number, + to=normalized, + ) + if normalized.startswith("uuid:"): + return cls( + peer=normalized, + chat_type=ChatType.DIRECT, + from_=account_number, + to=normalized, + ) + return cls( + peer=normalized, + chat_type=ChatType.DIRECT, + from_=account_number, + to=normalized, + ) + + @property + def is_group(self) -> bool: + return self.chat_type == ChatType.GROUP + + @property + def conversation_id(self) -> str: + return self.peer diff --git a/backend/package/yuxi/channels/adapters/signal/probe.py b/backend/package/yuxi/channels/adapters/signal/probe.py new file mode 100644 index 00000000..dc023f64 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/probe.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass, field + +import aiohttp + +from yuxi.channels.models import HealthStatus +from yuxi.channels.adapters.signal.client import RpcClient, RpcError + +logger = logging.getLogger(__name__) + +PROBE_TIMEOUT_MS = 7500 + + +class ProbeError(Exception): + def __init__(self, message: str, error_type: str = "UNKNOWN"): + super().__init__(message) + self.error_type = error_type + + +PROBE_ERROR_UNAUTHORIZED = "UNAUTHORIZED" +PROBE_ERROR_NOT_FOUND = "NOT_FOUND" +PROBE_ERROR_DAEMON_UNREACHABLE = "SIGNAL_DAEMON_UNREACHABLE" +PROBE_ERROR_TIMEOUT = "TIMEOUT" +PROBE_ERROR_UNKNOWN = "UNKNOWN" + + +@dataclass +class SignalProbeResult: + status: str = "unknown" + version: str | None = None + error_type: str | None = None + latency_ms: float = 0.0 + metadata: dict = field(default_factory=dict) + + @property + def success(self) -> bool: + return self.status == "healthy" and self.version is not None + + +async def probe_signal_daemon(rpc_client: RpcClient) -> SignalProbeResult: + import time as _time + + start = _time.monotonic() + base_url = rpc_client.base_url + + try: + about_result = await _probe_about_endpoint(base_url) + if about_result: + elapsed = (_time.monotonic() - start) * 1000 + return SignalProbeResult( + status="healthy", + version=about_result.get("version", "unknown"), + latency_ms=elapsed, + metadata={"arm": "about", "about": about_result}, + ) + except Exception: + logger.debug("About endpoint probe failed, falling back to version RPC") + + try: + version_result = await asyncio.wait_for( + rpc_client.call("version"), + timeout=PROBE_TIMEOUT_MS / 1000.0, + ) + elapsed = (_time.monotonic() - start) * 1000 + return SignalProbeResult( + status="healthy", + version=version_result.get("version", "unknown"), + latency_ms=elapsed, + metadata={"arm": "rpc_version"}, + ) + except TimeoutError: + return SignalProbeResult( + status="unhealthy", + error_type=PROBE_ERROR_TIMEOUT, + latency_ms=PROBE_TIMEOUT_MS, + ) + except RpcError as e: + return SignalProbeResult( + status="unhealthy", + error_type=_classify_rpc_error(e), + metadata={"error": str(e)}, + ) + except Exception as e: + return SignalProbeResult( + status="unhealthy", + error_type=PROBE_ERROR_DAEMON_UNREACHABLE, + metadata={"error": str(e)}, + ) + + +async def _probe_about_endpoint(base_url: str) -> dict | None: + url = f"{base_url}/api/v1/about" + try: + async with aiohttp.ClientSession() as session: + async with session.get(url) as resp: + if resp.status == 200: + return await resp.json() + except Exception: + pass + return None + + +async def health_check_signal(rpc_client: RpcClient) -> HealthStatus: + try: + probe_result = await probe_signal_daemon(rpc_client) + if not probe_result.success: + return HealthStatus( + status="unhealthy", + last_error=probe_result.error_type or "unknown error", + metadata=probe_result.metadata, + ) + return HealthStatus( + status="healthy", + metadata={ + "version": probe_result.version or "unknown", + "latency_ms": probe_result.latency_ms, + **probe_result.metadata, + }, + ) + except Exception as e: + return HealthStatus(status="unhealthy", last_error=str(e)) + + +def _classify_rpc_error(error: RpcError) -> str: + msg = str(error).lower() + if error.code == -32602: + return PROBE_ERROR_NOT_FOUND + if "unauthorized" in msg or "authorization" in msg or "forbidden" in msg: + return PROBE_ERROR_UNAUTHORIZED + if "not found" in msg or "missing" in msg: + return PROBE_ERROR_NOT_FOUND + if "refused" in msg or "unreachable" in msg or "connect" in msg: + return PROBE_ERROR_DAEMON_UNREACHABLE + return PROBE_ERROR_UNKNOWN diff --git a/backend/package/yuxi/channels/adapters/signal/reaction_level.py b/backend/package/yuxi/channels/adapters/signal/reaction_level.py new file mode 100644 index 00000000..a5e2622c --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/reaction_level.py @@ -0,0 +1,40 @@ +from enum import StrEnum + + +class ReactionLevel(StrEnum): + OFF = "off" + ACK = "ack" + MINIMAL = "minimal" + EXTENSIVE = "extensive" + + +class ReactionLevelController: + ACK_EMOJI = "\U0001f440" + + def __init__(self, level: str = "minimal"): + self.level = ReactionLevel(level) + self._ack_sent: set[str] = set() + + def should_send_reaction(self, chat_id: str, emoji: str | None = None) -> bool: + match self.level: + case ReactionLevel.OFF: + return False + case ReactionLevel.ACK: + return emoji == self.ACK_EMOJI + case ReactionLevel.MINIMAL: + return True + case ReactionLevel.EXTENSIVE: + return True + case _: + return True + + def should_send_auto_ack(self, chat_id: str) -> bool: + if self.level == ReactionLevel.OFF: + return False + if chat_id in self._ack_sent: + return False + self._ack_sent.add(chat_id) + return True + + def reset_ack(self, chat_id: str) -> None: + self._ack_sent.discard(chat_id) diff --git a/backend/package/yuxi/channels/adapters/signal/rpc_context.py b/backend/package/yuxi/channels/adapters/signal/rpc_context.py new file mode 100644 index 00000000..f284fdd8 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/rpc_context.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from yuxi.channels.adapters.signal.client import RpcClient + + +@dataclass +class RpcContext: + client: RpcClient + base_url: str + account: str + + @classmethod + async def resolve(cls, base_url: str, account: str) -> RpcContext: + client = RpcClient(base_url) + await client.connect() + return cls(client=client, base_url=base_url, account=account) + + async def disconnect(self) -> None: + await self.client.disconnect() diff --git a/backend/package/yuxi/channels/adapters/signal/security.py b/backend/package/yuxi/channels/adapters/signal/security.py new file mode 100644 index 00000000..8aae064b --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/security.py @@ -0,0 +1,214 @@ +from enum import StrEnum +import logging +from collections.abc import Awaitable, Callable + +from yuxi.channels.models import ChannelMessage + +logger = logging.getLogger(__name__) + + +class DmPolicy(StrEnum): + PAIRING = "pairing" + ALLOWLIST = "allowlist" + OPEN = "open" + DISABLED = "disabled" + + +class GroupPolicy(StrEnum): + OPEN = "open" + ALLOWLIST = "allowlist" + DISABLED = "disabled" + + +class ReactionNotificationPolicy(StrEnum): + OFF = "off" + OWN = "own" + ALLOWLIST = "allowlist" + ALL = "all" + + +class SignalSecurityPolicy: + def __init__( + self, + dm_policy: str = "pairing", + group_policy: str = "allowlist", + allow_from: list[str] | None = None, + group_allow_from: list[str] | None = None, + require_mention: bool = False, + reaction_notifications: str = "all", + reaction_allowlist: list[str] | None = None, + command_double_auth: bool = True, + ): + self.dm_policy = DmPolicy(dm_policy) + self.group_policy = GroupPolicy(group_policy) + self._dm_allowlist: set[str] = set(self._expand_allowlist(allow_from or [])) + self._group_allowlist: set[str] = set(group_allow_from or []) + self._pairing_pending: set[str] = set() + self.require_mention = require_mention + self.reaction_notifications = ReactionNotificationPolicy(reaction_notifications) + self._reaction_allowlist: set[str] = set(reaction_allowlist or []) + self._pairing_challenge_pending: set[str] = set() + self._command_double_auth = command_double_auth + self._store_write_fn: Callable[[str, dict], Awaitable[None]] | None = None + self._store_read_fn: Callable[[str], Awaitable[dict | None]] | None = None + + @staticmethod + def _expand_allowlist(entries: list[str]) -> list[str]: + result = [] + for entry in entries: + stripped = entry.strip() + if stripped.startswith("signal:"): + stripped = stripped.removeprefix("signal:") + result.append(stripped) + return result + + def _is_wildcard_match(self, user_id: str) -> bool: + return "*" in self._dm_allowlist + + def check_dm_permission(self, message: ChannelMessage) -> bool: + user_id = message.identity.channel_user_id + + match self.dm_policy: + case DmPolicy.DISABLED: + logger.info(f"[Signal Security] DM denied (policy=disabled, user={user_id})") + return False + case DmPolicy.OPEN: + return True + case DmPolicy.ALLOWLIST: + allowed = user_id in self._dm_allowlist or self._is_wildcard_match(user_id) + if not allowed: + logger.info(f"[Signal Security] DM denied (policy=allowlist, user={user_id})") + return allowed + case DmPolicy.PAIRING: + if user_id in self._dm_allowlist or self._is_wildcard_match(user_id): + return True + self._pairing_pending.add(user_id) + logger.info(f"[Signal Security] DM denied (policy=pairing, user={user_id}, pending)") + return False + case _: + logger.info(f"[Signal Security] DM denied (policy=unknown, user={user_id})") + return False + + def check_group_permission(self, message: ChannelMessage) -> bool: + group_id = message.identity.channel_chat_id + + match self.group_policy: + case GroupPolicy.DISABLED: + logger.info(f"[Signal Security] Group denied (policy=disabled, group={group_id})") + return False + case GroupPolicy.OPEN: + return True + case GroupPolicy.ALLOWLIST: + allowed = group_id in self._group_allowlist + if not allowed: + logger.info(f"[Signal Security] Group denied (policy=allowlist, group={group_id})") + return allowed + case _: + logger.info(f"[Signal Security] Group denied (policy=unknown, group={group_id})") + return False + + def check_require_mention(self, message: ChannelMessage) -> bool: + if not self.require_mention: + return True + if message.chat_type.value != "group": + return True + if message.mentions and message.mentions.is_bot_mentioned: + return True + return False + + def check_reaction_notification(self, user_id: str) -> bool: + match self.reaction_notifications: + case ReactionNotificationPolicy.OFF: + return False + case ReactionNotificationPolicy.OWN: + return False + case ReactionNotificationPolicy.ALLOWLIST: + return user_id in self._reaction_allowlist + case ReactionNotificationPolicy.ALL: + return True + case _: + return True + + def record_pairing_challenge(self, user_id: str) -> None: + self._pairing_challenge_pending.add(user_id) + + def has_pairing_challenge(self, user_id: str) -> bool: + return user_id in self._pairing_challenge_pending + + def approve_pairing(self, user_id: str) -> None: + self._dm_allowlist.add(user_id) + self._pairing_pending.discard(user_id) + + def reject_pairing(self, user_id: str) -> None: + self._pairing_pending.discard(user_id) + + @property + def pending_pairings(self) -> set[str]: + return self._pairing_pending.copy() + + def add_to_allowlist(self, target_id: str, target_type: str = "dm") -> None: + if target_type == "dm": + self._dm_allowlist.add(target_id) + elif target_type == "group": + self._group_allowlist.add(target_id) + else: + raise ValueError(f"Unknown target_type: {target_type}, expected 'dm' or 'group'") + + def remove_from_allowlist(self, target_id: str, target_type: str = "dm") -> None: + if target_type == "dm": + self._dm_allowlist.discard(target_id) + elif target_type == "group": + self._group_allowlist.discard(target_id) + else: + raise ValueError(f"Unknown target_type: {target_type}, expected 'dm' or 'group'") + + def check_command_double_auth(self, message: ChannelMessage) -> bool: + if not self._command_double_auth: + return True + user_id = message.identity.channel_user_id + group_id = message.identity.channel_chat_id + + dm_allowed = user_id in self._dm_allowlist or self._is_wildcard_match(user_id) + group_allowed = group_id in self._group_allowlist + + if dm_allowed or group_allowed: + return True + + logger.info( + f"[Signal Security] Command double-auth denied: user={user_id}, " + f"group={group_id}, dm_allowed={dm_allowed}, group_allowed={group_allowed}" + ) + return False + + def set_store_handlers( + self, + write_fn: Callable[[str, dict], Awaitable[None]], + read_fn: Callable[[str], Awaitable[dict | None]], + ) -> None: + self._store_write_fn = write_fn + self._store_read_fn = read_fn + + async def approve_pairing(self, user_id: str) -> None: + self._dm_allowlist.add(user_id) + self._pairing_pending.discard(user_id) + if self._store_write_fn: + try: + await self._store_write_fn( + f"signal:pairing:dm:{user_id}", + {"user_id": user_id, "approved_at": __import__("time").time()}, + ) + except Exception: + logger.exception("Failed to persist pairing approval to store") + + async def load_pairing_store(self) -> None: + if not self._store_read_fn: + return + try: + data = await self._store_read_fn("signal:pairing:dm:*") + if data and isinstance(data, dict): + for key, value in data.items(): + user_id = value.get("user_id") if isinstance(value, dict) else str(value) + if user_id: + self._dm_allowlist.add(user_id) + except Exception: + logger.exception("Failed to load pairing store") diff --git a/backend/package/yuxi/channels/adapters/signal/send.py b/backend/package/yuxi/channels/adapters/signal/send.py new file mode 100644 index 00000000..19c3c7af --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/send.py @@ -0,0 +1,428 @@ +from __future__ import annotations + +import asyncio +import base64 +import logging +import uuid + +from yuxi.channels.models import DeliveryResult +from yuxi.channels.adapters.signal.client import RpcClient, RpcError +from yuxi.channels.adapters.signal.format import ( + FormattedText, + StyleRange, + clamp_styles_to_length, + split_text, +) +from yuxi.channels.adapters.signal.normalize import normalize_target + +logger = logging.getLogger(__name__) + +GROUP_PREFIX = "group:" + + +def _make_message_id(timestamp: int | str | None) -> str | None: + if not timestamp: + return None + short_uuid = uuid.uuid4().hex[:8] + return f"{timestamp}:{short_uuid}" + + +class SignalSender: + MAX_TEXT_LENGTH = 4000 + MAX_RETRIES = 3 + RETRY_BASE_DELAY = 0.5 + CACHE_TTL = 3600 + CACHE_MAX_SIZE = 1000 + _sent_message_cache: dict[str, dict] = {} + _cache_access_order: list[str] = [] + + def __init__(self, rpc_client: RpcClient, account_number: str): + self._rpc = rpc_client + self._account = account_number + + @staticmethod + def _recipient_param(target: str) -> dict: + normalized = normalize_target(target) + if normalized.startswith(GROUP_PREFIX): + return {"groupId": normalized} + return {"recipient": normalized} + + def _cache_sent(self, msg_id: str, recipient: str) -> None: + self._prune_cache() + if len(self._sent_message_cache) >= self.CACHE_MAX_SIZE: + oldest = self._cache_access_order.pop(0) + self._sent_message_cache.pop(oldest, None) + self._sent_message_cache[msg_id] = { + "recipient": normalize_target(recipient), + "timestamp": asyncio.get_event_loop().time(), + } + self._cache_access_order.append(msg_id) + + def _get_cached(self, msg_id: str) -> dict | None: + entry = self._sent_message_cache.get(msg_id) + if entry is None: + return None + now = asyncio.get_event_loop().time() + if now - entry["timestamp"] > self.CACHE_TTL: + self._sent_message_cache.pop(msg_id, None) + if msg_id in self._cache_access_order: + self._cache_access_order.remove(msg_id) + return None + if msg_id in self._cache_access_order: + self._cache_access_order.remove(msg_id) + self._cache_access_order.append(msg_id) + return entry + + @classmethod + def _prune_cache(cls) -> None: + now = asyncio.get_event_loop().time() + expired = [k for k, v in cls._sent_message_cache.items() if now - v["timestamp"] > cls.CACHE_TTL] + for k in expired: + cls._sent_message_cache.pop(k, None) + if k in cls._cache_access_order: + cls._cache_access_order.remove(k) + + async def _call_with_retry(self, method: str, params: dict) -> dict: + last_error = None + for attempt in range(self.MAX_RETRIES): + try: + return await self._rpc.call(method, params) + except RpcError: + raise + except Exception as e: + last_error = e + if attempt < self.MAX_RETRIES - 1: + delay = self.RETRY_BASE_DELAY * (2**attempt) + await asyncio.sleep(delay) + raise last_error # type: ignore[misc] + + async def send_text( + self, + recipient: str, + message_body: str, + reply_to_id: str | None = None, + formatted_body: FormattedText | None = None, + chunk_mode: str = "newline", + text_mode: str = "markdown", + ) -> DeliveryResult: + if text_mode == "plain" and formatted_body is not None: + return await self._send_formatted_text(recipient, formatted_body, reply_to_id, chunk_mode) + + if formatted_body is not None and text_mode == "markdown": + return await self._send_formatted_text(recipient, formatted_body, reply_to_id, chunk_mode) + + chunks = split_text(message_body, self.MAX_TEXT_LENGTH, chunk_mode) + last_message_id = None + + for i, chunk in enumerate(chunks): + params = { + "account": self._account, + **self._recipient_param(recipient), + "messageBody": chunk, + } + if reply_to_id and i == 0: + params["quoteTimestamp"] = int(reply_to_id) + + try: + result = await self._call_with_retry("send", params) + last_message_id = _make_message_id(result.get("timestamp")) + if last_message_id: + self._cache_sent(last_message_id, recipient) + except Exception as e: + logger.error(f"Failed to send message to {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + return DeliveryResult(success=True, message_id=last_message_id) + + async def _send_formatted_text( + self, + recipient: str, + formatted: FormattedText, + reply_to_id: str | None = None, + chunk_mode: str = "newline", + ) -> DeliveryResult: + chunks = split_text(formatted.body, self.MAX_TEXT_LENGTH, chunk_mode) + last_message_id = None + offset = 0 + + for i, chunk in enumerate(chunks): + chunk_styles = _extract_chunk_styles(formatted.styles, offset, len(chunk)) + clamped = clamp_styles_to_length(chunk_styles, chunk) + + params: dict = { + "account": self._account, + **self._recipient_param(recipient), + "messageBody": chunk, + } + if clamped: + params["styledBody"] = chunk + params["styles"] = [{"start": s.start, "length": s.length, "style": s.style} for s in clamped] + if reply_to_id and i == 0: + params["quoteTimestamp"] = int(reply_to_id) + + try: + result = await self._call_with_retry("send", params) + last_message_id = _make_message_id(result.get("timestamp")) + if last_message_id: + self._cache_sent(last_message_id, recipient) + except Exception as e: + logger.error(f"Failed to send formatted message to {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + offset += len(chunk) + + return DeliveryResult(success=True, message_id=last_message_id) + + async def send_media( + self, + recipient: str, + media_data: bytes, + media_type: str, + filename: str | None = None, + caption: str | None = None, + ) -> DeliveryResult: + content_type_map = { + "image": "image/jpeg", + "video": "video/mp4", + "audio": "audio/ogg", + "file": "application/octet-stream", + } + + params = { + "account": self._account, + **self._recipient_param(recipient), + "messageBody": caption or "", + "attachments": [ + { + "contentType": content_type_map.get(media_type, "application/octet-stream"), + "filename": filename or "attachment", + "data": base64.b64encode(media_data).decode("utf-8"), + } + ], + } + + try: + result = await self._rpc.call("send", params) + return DeliveryResult( + success=True, + message_id=_make_message_id(result.get("timestamp")), + ) + except Exception as e: + logger.error(f"Failed to send media to {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def send_reaction( + self, + recipient: str, + target_author: str, + target_sent_timestamp: int, + reaction: str, + remove: bool = False, + ) -> DeliveryResult: + method = "removeReaction" if remove else "sendReaction" + params = { + "account": self._account, + **self._recipient_param(recipient), + "targetAuthor": normalize_target(target_author), + "targetSentTimestamp": target_sent_timestamp, + "reaction": reaction, + } + + try: + result = await self._rpc.call(method, params) + return DeliveryResult( + success=True, + message_id=_make_message_id(result.get("timestamp")), + ) + except Exception as e: + logger.error(f"Failed to send reaction to {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def send_typing_indicator(self, recipient: str) -> DeliveryResult: + try: + await self._rpc.call( + "sendTyping", + { + "account": self._account, + **self._recipient_param(recipient), + }, + ) + return DeliveryResult(success=True) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_read_receipt(self, recipient: str, timestamps: list[int]) -> DeliveryResult: + try: + await self._rpc.call( + "sendReadReceipt", + { + "account": self._account, + **self._recipient_param(recipient), + "timestamps": timestamps, + }, + ) + return DeliveryResult(success=True) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def edit_message( + self, + recipient: str, + target_author: str, + target_sent_timestamp: int, + new_body: str, + ) -> DeliveryResult: + try: + result = await self._rpc.call( + "editMessage", + { + "account": self._account, + **self._recipient_param(recipient), + "targetAuthor": normalize_target(target_author), + "targetSentTimestamp": target_sent_timestamp, + "newMessageBody": new_body, + }, + ) + return DeliveryResult( + success=True, + message_id=_make_message_id(result.get("timestamp")), + ) + except Exception as e: + logger.warning(f"editMessage failed for {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def delete_message(self, recipient: str, timestamps: list[int]) -> DeliveryResult: + try: + await self._rpc.call( + "remoteDelete", + { + "account": self._account, + **self._recipient_param(recipient), + "timestamps": timestamps, + }, + ) + return DeliveryResult(success=True) + except Exception as e: + logger.error(f"Failed to delete message for {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def send_sticker( + self, + recipient: str, + sticker_pack_id: str, + sticker_id: int, + ) -> DeliveryResult: + try: + params = { + "account": self._account, + **self._recipient_param(recipient), + "stickerPackId": sticker_pack_id, + "stickerId": sticker_id, + } + result = await self._rpc.call("sendSticker", params) + return DeliveryResult( + success=True, + message_id=_make_message_id(result.get("timestamp")), + ) + except Exception as e: + logger.error(f"Failed to send sticker to {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def send_silent_message( + self, + recipient: str, + message_body: str, + reply_to_id: str | None = None, + ) -> DeliveryResult: + try: + params = { + "account": self._account, + **self._recipient_param(recipient), + "messageBody": message_body, + "disableNotification": True, + } + if reply_to_id: + params["quoteTimestamp"] = int(reply_to_id) + result = await self._rpc.call("send", params) + return DeliveryResult( + success=True, + message_id=_make_message_id(result.get("timestamp")), + ) + except Exception as e: + logger.error(f"Failed to send silent message to {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def pin_message( + self, + recipient: str, + message_timestamp: int, + ) -> DeliveryResult: + try: + await self._rpc.call( + "pinMessage", + { + "account": self._account, + **self._recipient_param(recipient), + "targetSentTimestamp": message_timestamp, + }, + ) + return DeliveryResult(success=True) + except Exception as e: + logger.error(f"Failed to pin message for {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def unpin_message(self, recipient: str) -> DeliveryResult: + try: + await self._rpc.call( + "unpinMessage", + { + "account": self._account, + **self._recipient_param(recipient), + }, + ) + return DeliveryResult(success=True) + except Exception as e: + logger.error(f"Failed to unpin message for {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + async def send_voice( + self, + recipient: str, + audio_data: bytes, + duration_ms: int = 0, + ) -> DeliveryResult: + try: + params = { + "account": self._account, + **self._recipient_param(recipient), + "messageBody": "", + "attachments": [ + { + "contentType": "audio/ogg", + "filename": "voice.ogg", + "data": base64.b64encode(audio_data).decode("utf-8"), + } + ], + } + result = await self._rpc.call("send", params) + return DeliveryResult( + success=True, + message_id=_make_message_id(result.get("timestamp")), + ) + except Exception as e: + logger.error(f"Failed to send voice to {recipient}: {e}") + return DeliveryResult(success=False, error=str(e)) + + +def _extract_chunk_styles(styles: list[StyleRange], offset: int, chunk_len: int) -> list[StyleRange]: + chunk_end = offset + chunk_len + result: list[StyleRange] = [] + for s in styles: + s_end = s.start + s.length + if s_end <= offset or s.start >= chunk_end: + continue + new_start = max(0, s.start - offset) + new_end = min(chunk_len, s_end - offset) + if new_end > new_start: + result.append(StyleRange(start=new_start, length=new_end - new_start, style=s.style)) + return result diff --git a/backend/package/yuxi/channels/adapters/signal/session.py b/backend/package/yuxi/channels/adapters/signal/session.py new file mode 100644 index 00000000..2c0a6294 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/session.py @@ -0,0 +1,17 @@ +from yuxi.channels.models import ChannelIdentity, ChannelType +from yuxi.channels.adapters.signal.normalize import normalize_e164 + + +def resolve_thread(identity: ChannelIdentity, agent_id: str = "main") -> str: + chat_id = identity.channel_chat_id + channel_type = identity.channel_type + + if channel_type != ChannelType.SIGNAL: + raise ValueError(f"Unexpected channel type: {channel_type}") + + if chat_id.startswith("group:"): + group_id = chat_id.removeprefix("group:") + return f"agent:{agent_id}:signal:group:{group_id}" + + normalized = normalize_e164(chat_id) + return f"agent:{agent_id}:signal:dm:{normalized}" diff --git a/backend/package/yuxi/channels/adapters/signal/setup.py b/backend/package/yuxi/channels/adapters/signal/setup.py new file mode 100644 index 00000000..ab19824e --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/setup.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import re +from enum import StrEnum +from typing import Any + +E164_MIN_LENGTH = 5 +E164_MAX_LENGTH = 15 +E164_PATTERN = re.compile(r"^\+\d{%d,%d}$" % (E164_MIN_LENGTH, E164_MAX_LENGTH)) + + +class SetupStep(StrEnum): + STATUS = "status" + PREPARE = "prepare" + CLI_PATH = "cli_path" + SIGNAL_NUMBER = "signal_number" + ALLOW_FROM = "allow_from" + COMPLETION = "completion" + + +class SetupWizard: + def __init__(self, config: dict[str, Any] | None = None): + self.config = config or {} + self._current_step: SetupStep = SetupStep.STATUS + + @property + def current_step(self) -> str: + return self._current_step.value + + def get_step_result(self) -> dict: + match self._current_step: + case SetupStep.STATUS: + return self._status_step() + case SetupStep.PREPARE: + return self._prepare_step() + case SetupStep.CLI_PATH: + return self._cli_path_step() + case SetupStep.SIGNAL_NUMBER: + return self._signal_number_step() + case SetupStep.ALLOW_FROM: + return self._allow_from_step() + case SetupStep.COMPLETION: + return self._completion_step() + case _: + return {"step": self._current_step.value, "status": "unknown"} + + def advance(self, data: dict[str, Any] | None = None) -> dict: + step_order = list(SetupStep) + current_idx = step_order.index(self._current_step) + if current_idx < len(step_order) - 1: + self._current_step = step_order[current_idx + 1] + return self.get_step_result() + + def _status_step(self) -> dict: + return { + "step": "status", + "title": "Signal Setup Wizard", + "description": "Configure your Signal channel step by step", + "total_steps": len(SetupStep), + } + + def _prepare_step(self) -> dict: + from yuxi.channels.adapters.signal.install import check_java_installed, check_signal_cli_installed + + java_ok, _ = check_java_installed() + cli_ok, cli_version = check_signal_cli_installed(self.config.get("cli_path", "signal-cli")) + + result = { + "step": "prepare", + "title": "Environment Check", + "fields": [ + {"key": "java_installed", "label": "Java 17+", "ok": java_ok}, + {"key": "signal_cli_installed", "label": "signal-cli", "ok": cli_ok, "version": cli_version or ""}, + ], + "ready": java_ok and cli_ok, + } + + if not cli_ok and java_ok: + result["can_auto_install"] = True + result["auto_install_hint"] = "signal-cli can be auto-installed in the next step" + + return result + + def _cli_path_step(self) -> dict: + current_path = self.config.get("cli_path", "signal-cli") + return { + "step": "cli_path", + "title": "signal-cli Path", + "description": "Path to the signal-cli binary", + "fields": [ + { + "key": "cli_path", + "label": "Binary Path", + "type": "string", + "default": current_path, + "required": True, + } + ], + } + + def _signal_number_step(self) -> dict: + current_number = self.config.get("signal_number", "") + return { + "step": "signal_number", + "title": "Signal Phone Number", + "description": "E.164 format phone number (e.g. +1234567890)", + "fields": [ + { + "key": "signal_number", + "label": "Phone Number", + "type": "string", + "default": current_number, + "required": True, + "pattern": f"^\\+\\d{{{E164_MIN_LENGTH},{E164_MAX_LENGTH}}}$", + } + ], + } + + def _completion_step(self) -> dict: + return { + "step": "completion", + "title": "Setup Complete", + "description": "Signal channel is ready to connect", + "next_steps": [ + "1. Register your number: signal-cli -a register", + "2. Verify: signal-cli -a verify ", + "3. Add an account_uuid config for loop prevention", + ], + } + + def _allow_from_step(self) -> dict: + security = self.config.get("security", {}) + current_allow_from = security.get("allow_from", []) + return { + "step": "allow_from", + "title": "Access Control - Allow From", + "description": "Comma-separated E.164 numbers, UUIDs, or '*' to allow DM from (leave empty for pairing mode)", + "fields": [ + { + "key": "allow_from", + "label": "Allowed Senders", + "type": "string", + "default": ", ".join(current_allow_from) if current_allow_from else "", + "placeholder": "+8613800138000, uuid:abc123, *", + } + ], + } + + def auto_install(self) -> dict: + import asyncio + from yuxi.channels.adapters.signal.install import auto_install_signal_cli + + try: + target_dir = self.config.get("install_dir") + success = asyncio.run(auto_install_signal_cli(target_dir)) + return { + "step": "prepare", + "auto_install_success": success, + "message": "signal-cli installed successfully" if success else "auto-install failed", + } + except Exception as e: + return { + "step": "prepare", + "auto_install_success": False, + "message": str(e), + } + + +def validate_e164(number: str) -> bool: + return bool(E164_PATTERN.match(number.strip())) + + +def format_e164(raw: str) -> str: + stripped = raw.strip() + if E164_PATTERN.match(stripped): + return stripped + digits = re.sub(r"[^\d]", "", stripped) + if E164_MIN_LENGTH <= len(digits) <= E164_MAX_LENGTH: + return f"+{digits}" + raise ValueError(f"Invalid E.164 number: {raw}") diff --git a/backend/package/yuxi/channels/adapters/signal/sse_reconnect.py b/backend/package/yuxi/channels/adapters/signal/sse_reconnect.py new file mode 100644 index 00000000..220f0b2f --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/sse_reconnect.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import asyncio +import logging +import random +from collections.abc import Callable, Awaitable + +import aiohttp + +logger = logging.getLogger(__name__) + +MAX_RECONNECT_DELAY = 60.0 +INITIAL_RECONNECT_DELAY = 1.0 +JITTER = 0.1 + + +async def sse_event_stream( + url: str, + params: dict, + on_event: Callable[[str], Awaitable[None]], +) -> None: + reconnect_delay = INITIAL_RECONNECT_DELAY + last_event_id: str | None = None + unauth_backoff = False + + while True: + try: + headers: dict[str, str] = {"Accept": "text/event-stream"} + if last_event_id: + headers["Last-Event-ID"] = last_event_id + + async with aiohttp.ClientSession() as session: + async with session.get(url, params=params, headers=headers) as response: + if response.status == 401: + unauth_backoff = True + logger.error("SSE connection received 401 Unauthorized, backing off for 30s") + await asyncio.sleep(30.0) + reconnect_delay = INITIAL_RECONNECT_DELAY + continue + + if response.status != 200: + logger.error(f"SSE connection failed: {response.status}") + await asyncio.sleep(reconnect_delay) + reconnect_delay = min(reconnect_delay * 2, MAX_RECONNECT_DELAY) + continue + + if unauth_backoff: + logger.info("SSE reconnected successfully after 401 backoff") + unauth_backoff = False + + reconnect_delay = INITIAL_RECONNECT_DELAY + + async for line in response.content: + line_text = line.decode("utf-8").strip() + + if line_text.startswith("id:"): + last_event_id = line_text.removeprefix("id:").strip() + continue + + if line_text.startswith("data:"): + event_data = line_text.removeprefix("data:").strip() + if event_data: + try: + await on_event(event_data) + except Exception: + logger.exception("Error handling SSE event") + + except (TimeoutError, aiohttp.ClientError) as e: + logger.warning(f"SSE connection lost: {e}, reconnecting in {reconnect_delay:.1f}s") + jitter_ms = reconnect_delay * JITTER * random.random() + await asyncio.sleep(reconnect_delay + jitter_ms) + reconnect_delay = min(reconnect_delay * 2, MAX_RECONNECT_DELAY) + except asyncio.CancelledError: + logger.info("SSE event stream cancelled") + break diff --git a/backend/package/yuxi/channels/adapters/signal/token_utils.py b/backend/package/yuxi/channels/adapters/signal/token_utils.py new file mode 100644 index 00000000..6ceb90dd --- /dev/null +++ b/backend/package/yuxi/channels/adapters/signal/token_utils.py @@ -0,0 +1,12 @@ +from __future__ import annotations + + +def ensure_oauth_prefix(token: str) -> str: + token = token.strip() + if token.startswith("oauth:"): + return token + return f"oauth:{token}" + + +def normalize_token(token: str) -> str: + return ensure_oauth_prefix(token)