kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(channels/signal): 新增Signal渠道适配器完整实现

Browse Source 
新增了完整的Signal渠道适配器实现，包含RPC客户端、守护进程管理、安全策略、消息处理、安装配置工具等全套功能，支持通过signal-cli与Signal网络进行通信，包含账户管理、消息收发、反应处理、媒体分析、健康检查等能力。
This commit is contained in:
Kris 2026-05-12 00:48:25 +08:00
parent 552aef767c
commit 8dc86766f1
28 changed files with 4747 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
backend/package/yuxi/channels/adapters/signal/__init__.py Normal file
View File

@ -0,0 +1,4 @@
from yuxi.channels.registry import register_builtin_adapter
from yuxi.channels.adapters.signal.channel import SignalChannel
register_builtin_adapter(SignalChannel)

67
backend/package/yuxi/channels/adapters/signal/accounts.py Normal file
View File

@ -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",
}

992
backend/package/yuxi/channels/adapters/signal/channel.py Normal file
View File

@ -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 "<media:image>"
if mime_type.startswith("video/"):
return "<media:video>"
if mime_type.startswith("audio/"):
return "<media:audio>"
return "<media:file>"
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

187
backend/package/yuxi/channels/adapters/signal/client.py Normal file
View File

@ -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."
)

279
backend/package/yuxi/channels/adapters/signal/config_schema.py Normal file
View File

@ -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",
},
}

269
backend/package/yuxi/channels/adapters/signal/config_ui_hints.py Normal file
View File

@ -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",
},
}

214
backend/package/yuxi/channels/adapters/signal/daemon.py Normal file
View File

@ -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]}")

63
backend/package/yuxi/channels/adapters/signal/directory.py Normal file
View File

@ -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()
)

68
backend/package/yuxi/channels/adapters/signal/entity_cache.py Normal file
View File

@ -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,
}

75
backend/package/yuxi/channels/adapters/signal/event_queue.py Normal file
View File

@ -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)

59
backend/package/yuxi/channels/adapters/signal/exec_auth.py Normal file
View File

@ -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]

292
backend/package/yuxi/channels/adapters/signal/format.py Normal file
View File

@ -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

41
backend/package/yuxi/channels/adapters/signal/identity.py Normal file
View File

@ -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

210
backend/package/yuxi/channels/adapters/signal/install.py Normal file
View File

@ -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 <CODE>")
print(" 3. Start daemon mode:")
print(" signal-cli -a +1234567890 daemon --http-listen 127.0.0.1:8080")

57
backend/package/yuxi/channels/adapters/signal/media_vision.py Normal file
View File

@ -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'}]"

83
backend/package/yuxi/channels/adapters/signal/message_actions.py Normal file
View File

@ -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

241
backend/package/yuxi/channels/adapters/signal/monitor.py Normal file
View File

@ -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

376
backend/package/yuxi/channels/adapters/signal/normalize.py Normal file
View File

@ -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

46
backend/package/yuxi/channels/adapters/signal/outbound_session.py Normal file
View File

@ -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

137
backend/package/yuxi/channels/adapters/signal/probe.py Normal file
View File

@ -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

40
backend/package/yuxi/channels/adapters/signal/reaction_level.py Normal file
View File

@ -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)

21
backend/package/yuxi/channels/adapters/signal/rpc_context.py Normal file
View File

@ -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()

214
backend/package/yuxi/channels/adapters/signal/security.py Normal file
View File

@ -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")

428
backend/package/yuxi/channels/adapters/signal/send.py Normal file
View File

@ -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

17
backend/package/yuxi/channels/adapters/signal/session.py Normal file
View File

@ -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}"

180
backend/package/yuxi/channels/adapters/signal/setup.py Normal file
View File

@ -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 <number> register",
"2. Verify: signal-cli -a <number> verify <CODE>",
"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}")

75
backend/package/yuxi/channels/adapters/signal/sse_reconnect.py Normal file
View File

@ -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

12
backend/package/yuxi/channels/adapters/signal/token_utils.py Normal file
View File

@ -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)