diff --git a/backend/package/yuxi/channel/extensions/synology_chat/__init__.py b/backend/package/yuxi/channel/extensions/synology_chat/__init__.py new file mode 100644 index 00000000..59fec743 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/__init__.py @@ -0,0 +1,468 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.synology_chat.accounts import ( + list_account_ids, + resolve_account as _resolve_account_from_config, +) +from yuxi.channel.extensions.synology_chat.client import ( + build_button_attachment, + send_attachment, + send_file_url, + send_message, +) +from yuxi.channel.extensions.synology_chat.dedupe import SynologyChatDedupe +from yuxi.channel.extensions.synology_chat.format import ( + convert_message_links, + markdown_to_native, + sanitize_text, +) +from yuxi.channel.extensions.synology_chat.security import collect_security_warnings +from yuxi.channel.extensions.synology_chat.session import CHANNEL_FORMAT_INSTRUCTIONS +from yuxi.channel.extensions.synology_chat.types import ACCOUNT_ID_DEFAULT +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +if TYPE_CHECKING: + pass + +logger = logging.getLogger(__name__) + + +def _warn_webhook_path_conflicts(accounts: list) -> None: + path_counts: dict[str, list[str]] = {} + for account in accounts: + path_counts.setdefault(account.webhook_path, []).append(account.account_id) + + for path, ids in path_counts.items(): + if len(ids) > 1: + logger.warning( + "Synology Chat: duplicate-webhook-path: accounts %s share webhookPath=%s", + ids, + path, + ) + + +def _warn_inherited_webhook_paths(accounts: list) -> None: + from yuxi.channel.extensions.synology_chat.types import ( + WEBHOOK_PATH_SOURCE_INHERITED, + ACCOUNT_ID_DEFAULT, + ) + + for account in accounts: + if ( + account.account_id != ACCOUNT_ID_DEFAULT + and account.webhook_path_source == WEBHOOK_PATH_SOURCE_INHERITED + and not account.dangerously_allow_inherited_webhook_path + ): + logger.warning( + "Synology Chat [%s]: inherited-shared-webhook-path: named account " + "inherits base webhookPath, set dangerouslyAllowInheritedWebhookPath=true to suppress", + account.account_id, + ) + + +class SynologyChatPlugin(BaseChannelPlugin): + id = "synology-chat" + name = "Synology Chat" + order = 110 + + def __init__(self): + super().__init__() + self._dedupe = SynologyChatDedupe() + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct", "group"], + message_types=["text", "file"], + reactions=False, + typing_indicator=False, + threads=False, + edit=False, + unsend=False, + reply=False, + media=True, + effects=False, + native_commands=False, + polls=False, + streaming=False, + block_streaming=False, + ) + + # ── ConfigProtocol ──────────────────────────────────── + + def list_account_ids(self, config: dict) -> list[str]: + return list_account_ids(config) + + async def resolve_account(self, account_id: str) -> dict: + from yuxi.channel.runtime.manager import gateway + + account = _resolve_account_from_config(gateway.global_config, account_id) + return { + "account_id": account.account_id, + "enabled": account.enabled, + "token": account.token, + "incoming_url": account.incoming_url, + "nas_host": account.nas_host, + "webhook_path": account.webhook_path, + "dm_policy": account.dm_policy, + "allowed_user_ids": account.allowed_user_ids, + "rate_limit_per_minute": account.rate_limit_per_minute, + "bot_name": account.bot_name, + "allow_insecure_ssl": account.allow_insecure_ssl, + "dangerously_allow_name_matching": account.dangerously_allow_name_matching, + "dangerously_allow_inherited_webhook_path": account.dangerously_allow_inherited_webhook_path, + } + + def is_configured(self, account: dict) -> bool: + return bool(account.get("token")) and bool(account.get("incoming_url")) + + def is_enabled(self, account: dict) -> bool: + return account.get("enabled", True) + + def describe_account(self, account: dict) -> dict: + return { + "account_id": account.get("account_id", ""), + "bot_name": account.get("bot_name", "ForcePilot"), + "nas_host": account.get("nas_host", ""), + "dm_policy": account.get("dm_policy", "allowlist"), + } + + def config_schema(self) -> dict: + return { + "type": "object", + "properties": { + "token": { + "type": "string", + "title": "Webhook Token", + "description": "Synology Chat 传出 Webhook 的 Token", + }, + "incomingUrl": { + "type": "string", + "title": "Incoming Webhook URL", + "description": "Synology Chat 传入 Webhook 的完整 URL", + }, + "nasHost": { + "type": "string", + "title": "NAS Host", + "description": "Synology NAS 主机地址", + "default": "localhost", + }, + "dmPolicy": { + "type": "string", + "title": "DM 策略", + "enum": ["open", "allowlist", "disabled"], + "default": "allowlist", + }, + "allowedUserIds": { + "type": "array", + "title": "允许的用户 ID", + "items": {"type": "string"}, + }, + "botName": { + "type": "string", + "title": "Bot 名称", + "default": "ForcePilot", + }, + "webhookPath": { + "type": "string", + "title": "Webhook 路径", + "default": "/webhook/synology", + }, + "rateLimitPerMinute": { + "type": "integer", + "title": "每分钟限速", + "default": 30, + }, + "allowInsecureSsl": { + "type": "boolean", + "title": "允许不安全 SSL", + "default": False, + }, + }, + "accounts": { + "type": "object", + "title": "多账户配置", + "additionalProperties": { + "type": "object", + "properties": { + "token": {"type": "string", "title": "Token"}, + "incomingUrl": {"type": "string", "title": "Incoming URL"}, + "nasHost": {"type": "string", "title": "NAS Host"}, + "webhookPath": {"type": "string", "title": "Webhook Path"}, + "botName": {"type": "string", "title": "Bot Name"}, + "enabled": {"type": "boolean", "title": "启用", "default": True}, + }, + }, + }, + } + + # ── GatewayProtocol ─────────────────────────────────── + + async def start(self, ctx) -> object: + from yuxi.channel.runtime.manager import gateway + + cfg = gateway.global_config or {} + channel_cfg = cfg.get("channels", {}).get("synology-chat", {}) + if not channel_cfg: + logger.info("Synology Chat not configured, skipping start") + return None + + account_ids = list_account_ids(cfg) + enabled_count = 0 + for aid in account_ids: + account = _resolve_account_from_config(cfg, aid) + if not account.enabled: + continue + enabled_count += 1 + + logger.info("Synology Chat started with %d enabled account(s)", enabled_count) + return None + + async def stop(self, ctx) -> None: + self._dedupe.reset() + logger.info("Synology Chat stopped") + + # ── StatusProtocol ──────────────────────────────────── + + async def probe(self, account: dict | None = None) -> bool: + if account is None: + from yuxi.channel.runtime.manager import gateway + + account = _resolve_account_from_config(gateway.global_config, ACCOUNT_ID_DEFAULT) + from yuxi.channel.extensions.synology_chat.status import probe as _probe + + return await _probe(account) + + def build_summary(self, snapshot: object) -> dict: + return {"channel": self.id, "name": self.name, "healthy": True} + + # ── OutboundProtocol ────────────────────────────────── + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + chat_type: str = "direct", + ) -> None: + from yuxi.channel.runtime.manager import gateway + + account_id = account_id or ACCOUNT_ID_DEFAULT + account = _resolve_account_from_config(gateway.global_config, account_id) + + if not account.incoming_url: + logger.warning("No incoming_url configured for account %s, cannot send message", account_id) + return + + if chat_type == "group": + await send_message( + incoming_url=account.incoming_url, + text=content, + channel_id=target_id, + allow_insecure_ssl=account.allow_insecure_ssl, + ) + else: + await send_message( + incoming_url=account.incoming_url, + text=content, + user_id=target_id, + allow_insecure_ssl=account.allow_insecure_ssl, + ) + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + from yuxi.channel.runtime.manager import gateway + + account_id = account_id or ACCOUNT_ID_DEFAULT + account = _resolve_account_from_config(gateway.global_config, account_id) + + if not account.incoming_url: + logger.warning("No incoming_url configured for account %s, cannot send media", account_id) + return + + await send_file_url( + incoming_url=account.incoming_url, + file_url=media_url, + user_id=target_id, + allow_insecure_ssl=account.allow_insecure_ssl, + ) + + async def send_interactive( + self, + target_id: str, + text: str, + callback_id: str, + buttons: list[dict], + *, + account_id: str | None = None, + ) -> bool: + from yuxi.channel.runtime.manager import gateway + + account_id = account_id or ACCOUNT_ID_DEFAULT + account = _resolve_account_from_config(gateway.global_config, account_id) + + if not account.incoming_url: + logger.warning("No incoming_url configured for account %s", account_id) + return False + + attachment = build_button_attachment(callback_id, text, buttons) + return await send_attachment( + incoming_url=account.incoming_url, + text=text, + attachments=[attachment], + user_id=target_id, + allow_insecure_ssl=account.allow_insecure_ssl, + ) + + # ── FormatProtocol ──────────────────────────────────── + + @property + def channel_format_instructions(self) -> str | None: + return CHANNEL_FORMAT_INSTRUCTIONS + + def sanitize_text(self, text: str, payload: object | None = None) -> str: + return sanitize_text(text, payload) + + def markdown_to_native(self, md_text: str) -> str: + return markdown_to_native(md_text) + + async def on_message_sending(self, msg: object, content: str) -> str | None: + return convert_message_links(content) + + # ── SecurityProtocol ────────────────────────────────── + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + from yuxi.channel.runtime.manager import gateway + from yuxi.channel.extensions.synology_chat.security import authorize_dm + + account = _resolve_account_from_config(gateway.global_config, ACCOUNT_ID_DEFAULT) + allowed, _ = authorize_dm(account, peer_id) + return allowed + + # ── GroupsProtocol ──────────────────────────────────── + + async def list_groups(self) -> list[dict]: + from yuxi.channel.runtime.manager import gateway + from yuxi.channel.extensions.synology_chat.client import fetch_chat_channels + + account = _resolve_account_from_config(gateway.global_config, ACCOUNT_ID_DEFAULT) + if not account.incoming_url: + return [] + + channels = await fetch_chat_channels( + account.incoming_url, + allow_insecure_ssl=account.allow_insecure_ssl, + ) + + return [ + { + "group_id": ch.get("channel_id", ""), + "name": ch.get("name", ch.get("channel_id", "")), + "members_count": ch.get("member_count", 0), + } + for ch in channels + ] + + # ── DedupeProtocol ──────────────────────────────────── + + def is_duplicate(self, key: str) -> bool: + return self._dedupe.is_duplicate(key) + + def mark_seen(self, key: str) -> None: + self._dedupe.mark_seen(key) + + def reset(self) -> None: + self._dedupe.reset() + + @property + def ttl_seconds(self) -> int: + return self._dedupe.ttl_seconds + + @property + def max_entries(self) -> int: + return self._dedupe.max_entries + + # ── AgentToolProtocol ───────────────────────────────── + + def get_agent_tools(self) -> list: + return [ + { + "name": "synology_list_users", + "description": "列出 Synology Chat 中的用户", + "parameters": [], + }, + ] + + async def execute_agent_tool(self, tool_name: str, params: dict, context: dict) -> dict: + from yuxi.channel.runtime.manager import gateway + from yuxi.channel.extensions.synology_chat.client import fetch_chat_users + + account = _resolve_account_from_config(gateway.global_config, ACCOUNT_ID_DEFAULT) + if not account.incoming_url: + return {"success": False, "error": "No incoming_url configured"} + + match tool_name: + case "synology_list_users": + users = await fetch_chat_users( + account.incoming_url, + allow_insecure_ssl=account.allow_insecure_ssl, + ) + return { + "success": True, + "result": [ + {"user_id": u.get("user_id"), "username": u.get("username"), "nickname": u.get("nickname")} + for u in users + ], + } + case _: + return {"success": False, "error": f"Unknown tool: {tool_name}"} + + # ── LifecycleProtocol ───────────────────────────────── + + async def run_startup_maintenance(self, cfg: dict) -> None: + channel_cfg = cfg.get("channels", {}).get("synology-chat", {}) + if not channel_cfg: + logger.info("Synology Chat not configured, skipping startup maintenance") + return + + account_ids = list_account_ids(cfg) + accounts = [] + for aid in account_ids: + account = _resolve_account_from_config(cfg, aid) + accounts.append(account) + if not account.enabled: + logger.info("Synology Chat account [%s] is disabled, skipping", aid) + continue + + warnings = collect_security_warnings(account) + for w in warnings: + logger.warning("Synology Chat [%s]: %s", account.account_id, w) + + if not account.token or not account.incoming_url: + logger.warning( + "Synology Chat [%s]: token or incomingUrl not configured, channel will not accept messages", + aid, + ) + + enabled_accounts = [a for a in accounts if a.enabled] + + _warn_webhook_path_conflicts(enabled_accounts) + _warn_inherited_webhook_paths(enabled_accounts) + + +ChannelPluginRegistry.register(SynologyChatPlugin()) diff --git a/backend/package/yuxi/channel/extensions/synology_chat/accounts.py b/backend/package/yuxi/channel/extensions/synology_chat/accounts.py new file mode 100644 index 00000000..bd9b13eb --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/accounts.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import os + +from yuxi.channel.extensions.synology_chat.types import ( + ACCOUNT_ID_DEFAULT, + BOT_NAME_DEFAULT, + RATE_LIMIT_DEFAULT, + WEBHOOK_PATH_DEFAULT, + WEBHOOK_PATH_SOURCE_DEFAULT, + WEBHOOK_PATH_SOURCE_EXPLICIT, + WEBHOOK_PATH_SOURCE_INHERITED, + DM_POLICY_ALLOWLIST, + ResolvedSynologyChatAccount, +) + + +def list_account_ids(config: dict) -> list[str]: + channel_cfg = config.get("channels", {}).get("synology-chat", {}) + base_ids = [ACCOUNT_ID_DEFAULT] + named = list(channel_cfg.get("accounts", {}).keys()) + return base_ids + named + + +def resolve_account(config: dict, account_id: str = ACCOUNT_ID_DEFAULT) -> ResolvedSynologyChatAccount: + channel_cfg = config.get("channels", {}).get("synology-chat", {}) + account_overrides = channel_cfg.get("accounts", {}).get(account_id, {}) if account_id != ACCOUNT_ID_DEFAULT else {} + + def _resolve(key: str, env_var: str | None = None, default: str | int | bool = "") -> str | int | bool: + if key in account_overrides: + return account_overrides[key] + if key in channel_cfg: + return channel_cfg[key] + if env_var and account_id == ACCOUNT_ID_DEFAULT: + env_val = os.environ.get(env_var, "") + if env_val: + return env_val + return default + + def _resolve_list(key: str) -> list[str]: + if key in account_overrides: + val = account_overrides[key] + return val if isinstance(val, list) else [] + if key in channel_cfg: + val = channel_cfg[key] + return val if isinstance(val, list) else [] + if account_id == ACCOUNT_ID_DEFAULT: + env_val = os.environ.get("SYNOLOGY_ALLOWED_USER_IDS", "") + if env_val: + return [u.strip() for u in env_val.split(",") if u.strip()] + return [] + + def _resolve_webhook_path_source() -> str: + if account_overrides.get("webhookPath"): + return WEBHOOK_PATH_SOURCE_EXPLICIT + if account_id != ACCOUNT_ID_DEFAULT and channel_cfg.get("webhookPath"): + return WEBHOOK_PATH_SOURCE_INHERITED + return WEBHOOK_PATH_SOURCE_DEFAULT + + return ResolvedSynologyChatAccount( + account_id=account_id, + enabled=bool(_resolve("enabled", default=True)), + token=str(_resolve("token", "SYNOLOGY_CHAT_TOKEN")), + incoming_url=str(_resolve("incomingUrl", "SYNOLOGY_CHAT_INCOMING_URL")), + nas_host=str(_resolve("nasHost", "SYNOLOGY_NAS_HOST", "localhost")), + webhook_path=str(_resolve("webhookPath", default=WEBHOOK_PATH_DEFAULT)), + webhook_path_source=_resolve_webhook_path_source(), + dm_policy=str(_resolve("dmPolicy", default=DM_POLICY_ALLOWLIST)), + allowed_user_ids=_resolve_list("allowedUserIds"), + rate_limit_per_minute=int(_resolve("rateLimitPerMinute", "SYNOLOGY_RATE_LIMIT", RATE_LIMIT_DEFAULT)), + bot_name=str(_resolve("botName", "OPENCLAW_BOT_NAME", BOT_NAME_DEFAULT)), + allow_insecure_ssl=bool(_resolve("allowInsecureSsl", default=False)), + dangerously_allow_name_matching=bool(_resolve("dangerouslyAllowNameMatching", default=False)), + dangerously_allow_inherited_webhook_path=bool(_resolve("dangerouslyAllowInheritedWebhookPath", default=False)), + ) diff --git a/backend/package/yuxi/channel/extensions/synology_chat/client.py b/backend/package/yuxi/channel/extensions/synology_chat/client.py new file mode 100644 index 00000000..1b1d54d4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/client.py @@ -0,0 +1,320 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import time +import urllib.parse +from urllib.parse import urlencode + +import aiohttp + +from yuxi.channel.extensions.synology_chat.types import ( + BASE_RETRY_DELAY_S, + CACHE_TTL_S, + HTTP_TIMEOUT_S, + MAX_RETRIES, + MIN_SEND_INTERVAL_MS, +) + +logger = logging.getLogger(__name__) + +_last_send_time: float = 0.0 +_chat_user_cache: dict[str, tuple[float, list[dict]]] = {} + + +async def send_message( + incoming_url: str, + text: str, + user_id: str | None = None, + channel_id: str | None = None, + allow_insecure_ssl: bool = False, +) -> bool: + global _last_send_time + + elapsed = (time.monotonic() - _last_send_time) * 1000 + if elapsed < MIN_SEND_INTERVAL_MS: + await asyncio.sleep((MIN_SEND_INTERVAL_MS - elapsed) / 1000) + + payload: dict = {"text": text} + if user_id: + payload["user_ids"] = [int(user_id)] + if channel_id: + payload["channel_id"] = channel_id + + body = f"payload={urlencode({'payload': json.dumps(payload)})}" + ssl_context: bool | None = False if allow_insecure_ssl else None + + for attempt in range(MAX_RETRIES): + try: + _last_send_time = time.monotonic() + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=HTTP_TIMEOUT_S)) as session: + async with session.post( + incoming_url, + data=body, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ssl=ssl_context if ssl_context is not None else True, + ) as resp: + if resp.status < 500: + return resp.status < 400 + except Exception: + logger.exception("Failed to send message (attempt %d/%d)", attempt + 1, MAX_RETRIES) + if attempt < MAX_RETRIES - 1: + delay = BASE_RETRY_DELAY_S * (2**attempt) + await asyncio.sleep(delay) + + return False + + +async def send_file_url( + incoming_url: str, + file_url: str, + user_id: str | None = None, + allow_insecure_ssl: bool = False, +) -> bool: + parsed = urllib.parse.urlparse(file_url) + if parsed.scheme not in ("http", "https"): + logger.warning("SSRF blocked: invalid scheme for file URL %s", file_url) + return False + + payload: dict = {"file_url": file_url} + if user_id: + payload["user_ids"] = [int(user_id)] + + body = f"payload={urlencode({'payload': json.dumps(payload)})}" + ssl_context: bool | None = False if allow_insecure_ssl else None + + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=HTTP_TIMEOUT_S)) as session: + async with session.post( + incoming_url, + data=body, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ssl=ssl_context if ssl_context is not None else True, + ) as resp: + return resp.status < 400 + + +def _build_user_list_url(incoming_url: str) -> str: + parsed = urllib.parse.urlparse(incoming_url) + base = f"{parsed.scheme}://{parsed.netloc}/webapi/entry.cgi" + params = { + "api": "SYNO.Chat.External", + "method": "user_list", + "version": "2", + } + return f"{base}?{urlencode(params)}" + + +async def fetch_chat_users( + incoming_url: str, + allow_insecure_ssl: bool = False, +) -> list[dict]: + list_url = _build_user_list_url(incoming_url) + + if list_url in _chat_user_cache: + cached_at, users = _chat_user_cache[list_url] + if time.monotonic() - cached_at < CACHE_TTL_S: + return users + + ssl_context: bool | None = False if allow_insecure_ssl else None + + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10)) as session: + async with session.get( + list_url, + ssl=ssl_context if ssl_context is not None else True, + ) as resp: + data = await resp.json() + users = data.get("data", {}).get("users", []) + + _chat_user_cache[list_url] = (time.monotonic(), users) + return users + + +async def send_attachment( + incoming_url: str, + text: str, + attachments: list[dict], + user_id: str | None = None, + allow_insecure_ssl: bool = False, +) -> bool: + global _last_send_time + + elapsed = (time.monotonic() - _last_send_time) * 1000 + if elapsed < MIN_SEND_INTERVAL_MS: + await asyncio.sleep((MIN_SEND_INTERVAL_MS - elapsed) / 1000) + + payload: dict = {"text": text, "attachments": attachments} + if user_id: + payload["user_ids"] = [int(user_id)] + + body = f"payload={urlencode({'payload': json.dumps(payload)})}" + ssl_context: bool | None = False if allow_insecure_ssl else None + + for attempt in range(MAX_RETRIES): + try: + _last_send_time = time.monotonic() + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=HTTP_TIMEOUT_S)) as session: + async with session.post( + incoming_url, + data=body, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ssl=ssl_context if ssl_context is not None else True, + ) as resp: + if resp.status < 500: + return resp.status < 400 + except Exception: + logger.exception("Failed to send attachment (attempt %d/%d)", attempt + 1, MAX_RETRIES) + if attempt < MAX_RETRIES - 1: + delay = BASE_RETRY_DELAY_S * (2**attempt) + await asyncio.sleep(delay) + + return False + + +def build_button_attachment( + callback_id: str, + text: str, + buttons: list[dict], +) -> dict: + actions = [] + for btn in buttons: + actions.append( + { + "type": "button", + "name": btn["name"], + "value": btn["value"], + "text": btn["text"], + "style": btn.get("style", "grey"), + } + ) + return { + "callback_id": callback_id, + "text": text, + "actions": actions, + } + + +def _build_channel_list_url(incoming_url: str) -> str: + parsed = urllib.parse.urlparse(incoming_url) + base = f"{parsed.scheme}://{parsed.netloc}/webapi/entry.cgi" + params = { + "api": "SYNO.Chat.External", + "method": "channel_list", + "version": "2", + } + return f"{base}?{urlencode(params)}" + + +_channel_list_cache: dict[str, tuple[float, list[dict]]] = {} + + +async def fetch_chat_channels( + incoming_url: str, + allow_insecure_ssl: bool = False, +) -> list[dict]: + list_url = _build_channel_list_url(incoming_url) + + if list_url in _channel_list_cache: + cached_at, channels = _channel_list_cache[list_url] + if time.monotonic() - cached_at < CACHE_TTL_S: + return channels + + ssl_context: bool | None = False if allow_insecure_ssl else None + + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10)) as session: + async with session.get( + list_url, + ssl=ssl_context if ssl_context is not None else True, + ) as resp: + data = await resp.json() + channels = data.get("data", {}).get("channels", []) + + _channel_list_cache[list_url] = (time.monotonic(), channels) + return channels + + +def _build_post_list_url(incoming_url: str, channel_id: str | None = None) -> str: + parsed = urllib.parse.urlparse(incoming_url) + base = f"{parsed.scheme}://{parsed.netloc}/webapi/entry.cgi" + params = { + "api": "SYNO.Chat.External", + "method": "post_list", + "version": "2", + } + if channel_id: + params["channel_id"] = channel_id + return f"{base}?{urlencode(params)}" + + +async def fetch_chat_posts( + incoming_url: str, + channel_id: str | None = None, + offset: int = 0, + limit: int = 50, + allow_insecure_ssl: bool = False, +) -> list[dict]: + list_url = _build_post_list_url(incoming_url, channel_id) + params = {"offset": offset, "limit": limit} + + ssl_context: bool | None = False if allow_insecure_ssl else None + + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15)) as session: + async with session.get( + list_url, + params=params, + ssl=ssl_context if ssl_context is not None else True, + ) as resp: + data = await resp.json() + return data.get("data", {}).get("posts", []) + + +def _build_file_get_url(incoming_url: str, file_id: str) -> str: + parsed = urllib.parse.urlparse(incoming_url) + base = f"{parsed.scheme}://{parsed.netloc}/webapi/entry.cgi" + params = { + "api": "SYNO.Chat.External", + "method": "post_file_get", + "version": "2", + "file_id": file_id, + } + return f"{base}?{urlencode(params)}" + + +async def fetch_chat_file_bytes( + incoming_url: str, + file_id: str, + allow_insecure_ssl: bool = False, +) -> bytes | None: + file_url = _build_file_get_url(incoming_url, file_id) + + ssl_context: bool | None = False if allow_insecure_ssl else None + + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as session: + async with session.get( + file_url, + ssl=ssl_context if ssl_context is not None else True, + ) as resp: + if resp.status != 200: + return None + return await resp.read() + + +async def resolve_legacy_webhook_name_to_chat_user_id( + incoming_url: str, + webhook_username: str, + allow_insecure_ssl: bool = False, +) -> str | None: + if not webhook_username: + return None + + users = await fetch_chat_users(incoming_url, allow_insecure_ssl) + + for user in users: + if user.get("nickname") == webhook_username: + return str(user.get("user_id")) + + for user in users: + if user.get("username") == webhook_username: + return str(user.get("user_id")) + + return None diff --git a/backend/package/yuxi/channel/extensions/synology_chat/dedupe.py b/backend/package/yuxi/channel/extensions/synology_chat/dedupe.py new file mode 100644 index 00000000..c7c9f39c --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/dedupe.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +import time +from collections import OrderedDict + + +class SynologyChatDedupe: + def __init__(self, ttl_seconds: int = 300, max_entries: int = 10000): + self._ttl_seconds = ttl_seconds + self._max_entries = max_entries + self._store: OrderedDict[str, float] = OrderedDict() + + def is_duplicate(self, key: str) -> bool: + now = time.monotonic() + if key in self._store: + ts = self._store[key] + if now - ts < self._ttl_seconds: + return True + del self._store[key] + return False + + def mark_seen(self, key: str) -> None: + now = time.monotonic() + if key in self._store: + self._store.move_to_end(key) + self._store[key] = now + self._evict() + + def _evict(self) -> None: + while len(self._store) > self._max_entries: + self._store.popitem(last=False) + + def reset(self) -> None: + self._store.clear() + + @property + def ttl_seconds(self) -> int: + return self._ttl_seconds + + @property + def max_entries(self) -> int: + return self._max_entries \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/synology_chat/format.py b/backend/package/yuxi/channel/extensions/synology_chat/format.py new file mode 100644 index 00000000..39c22ff5 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/format.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +import re + +_LINK_PATTERN = re.compile(r"\[([^\]]*)\]\(([^)]+)\)") + + +def sanitize_text(text: str, payload: object | None = None) -> str: + return text + + +def markdown_to_native(md_text: str) -> str: + def _convert(match): + label = match.group(1) + url = match.group(2) + if not label or not url: + return match.group(0) + return f"<{url}|{label}>" + + return _LINK_PATTERN.sub(_convert, md_text) + + +def convert_message_links(content: str) -> str: + return markdown_to_native(content) \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/synology_chat/plugin.json b/backend/package/yuxi/channel/extensions/synology_chat/plugin.json new file mode 100644 index 00000000..31bd0b76 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/plugin.json @@ -0,0 +1,28 @@ +{ + "id": "synology-chat", + "name": "Synology Chat", + "version": "1.0.0", + "label": "Synology Chat", + "description": "Synology Chat channel plugin for direct and group messaging via webhook with text and file support", + "author": "ForcePilot", + "order": 110, + "dependencies": ["aiohttp"], + "python_requires": ">=3.12", + "enabled": true, + "capabilities": { + "chat_types": ["direct", "group"], + "message_types": ["text", "file"], + "reactions": false, + "typing_indicator": false, + "threads": false, + "edit": false, + "unsend": false, + "reply": false, + "media": true, + "effects": false, + "native_commands": false, + "polls": false, + "streaming": false, + "block_streaming": false + } +} diff --git a/backend/package/yuxi/channel/extensions/synology_chat/security.py b/backend/package/yuxi/channel/extensions/synology_chat/security.py new file mode 100644 index 00000000..53a90def --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/security.py @@ -0,0 +1,170 @@ +from __future__ import annotations + +import hmac +import logging +import re +import time +from collections import OrderedDict +from dataclasses import dataclass + +from yuxi.channel.extensions.synology_chat.types import ( + INPUT_MAX_LENGTH, + DM_POLICY_ALLOWLIST, + DM_POLICY_DISABLED, + DM_POLICY_OPEN, + ResolvedSynologyChatAccount, +) + +logger = logging.getLogger(__name__) + +INVALID_TOKEN_WINDOW_MS = 60_000 +INVALID_TOKEN_MAX_TRACKED_KEYS = 5000 +INVALID_TOKEN_MAX_REQUESTS = 10 + +DANGEROUS_PATTERNS: list[tuple[re.Pattern, str]] = [ + (re.compile(r"ignore\s+(all\s+)?(previous|prior|above)\s+(instructions?|prompts?)", re.IGNORECASE), "[FILTERED]"), + (re.compile(r"you\s+are\s+now\s+", re.IGNORECASE), "[FILTERED]"), + (re.compile(r"system:\s*", re.IGNORECASE), "[FILTERED]"), + (re.compile(r"<\|.*?\|>"), "[FILTERED]"), +] + + +def validate_token(received: str, expected: str) -> bool: + if not received or not expected: + return False + return hmac.compare_digest(received.encode(), expected.encode()) + + +def authorize_dm(account: ResolvedSynologyChatAccount, user_id: str) -> tuple[bool, str]: + policy = account.dm_policy + allowed = account.allowed_user_ids + + if policy == DM_POLICY_DISABLED: + return False, "DM is disabled" + + if policy == DM_POLICY_OPEN: + if "*" in allowed: + return True, "" + if not allowed: + return False, "not-allowlisted" + return user_id in allowed, "not-allowlisted" + + if policy == DM_POLICY_ALLOWLIST: + if not allowed: + return False, "allowlist-empty" + return user_id in allowed, "not-allowlisted" + + return False, "unknown-policy" + + +def sanitize_input(text: str) -> str: + for pattern, replacement in DANGEROUS_PATTERNS: + text = pattern.sub(replacement, text) + + if len(text) > INPUT_MAX_LENGTH: + text = text[:INPUT_MAX_LENGTH] + " ... [truncated]" + + return text + + +def collect_security_warnings(account: ResolvedSynologyChatAccount) -> list[str]: + warnings: list[str] = [] + + if not account.token: + warnings.append("token is not configured. The webhook will reject all requests.") + + if not account.incoming_url: + warnings.append("incomingUrl is not configured. The bot cannot send replies.") + + if account.allow_insecure_ssl: + warnings.append("SSL verification is disabled. Only use for local NAS with self-signed certificates.") + + if account.dangerously_allow_name_matching: + warnings.append("dangerouslyAllowNameMatching re-enables mutable username/nickname recipient matching.") + + if account.dm_policy == DM_POLICY_OPEN and not account.allowed_user_ids: + warnings.append('Add allowedUserIds=["*"] for public DMs or set explicit user IDs.') + + if account.dm_policy == DM_POLICY_ALLOWLIST and not account.allowed_user_ids: + warnings.append('Add users or set dmPolicy="open" with allowedUserIds=["*"].') + + return warnings + + +@dataclass +class _WindowState: + count: int + window_start_ms: float + + +class InvalidTokenRateLimiter: + def __init__(self, max_requests: int = INVALID_TOKEN_MAX_REQUESTS): + self._max_requests = max_requests + self._state: OrderedDict[str, _WindowState] = OrderedDict() + + def allow(self, key: str) -> bool: + now = time.monotonic() * 1000 + state = self._state.get(key) + + if state is None or (now - state.window_start_ms) >= INVALID_TOKEN_WINDOW_MS: + self._state[key] = _WindowState(count=1, window_start_ms=now) + self._touch(key) + return True + + state.count += 1 + self._touch(key) + return state.count <= self._max_requests + + def _touch(self, key: str) -> None: + self._state.move_to_end(key) + while len(self._state) > INVALID_TOKEN_MAX_TRACKED_KEYS: + self._state.popitem(last=False) + + def clear(self) -> None: + self._state.clear() + + +class UserRateLimiter: + def __init__(self, max_per_window: int = 30, window_ms: int = 60_000): + self._max = max_per_window + self._window_ms = window_ms + self._state: dict[str, _WindowState] = {} + + def allow(self, user_id: str) -> bool: + now = time.monotonic() * 1000 + state = self._state.get(user_id) + + if state is None or (now - state.window_start_ms) >= self._window_ms: + self._state[user_id] = _WindowState(count=1, window_start_ms=now) + return True + + state.count += 1 + return state.count <= self._max + + def clear(self) -> None: + self._state.clear() + + def max_requests(self) -> int: + return self._max + + +_rate_limiters: dict[str, UserRateLimiter] = {} +_invalid_token_limiters: dict[str, InvalidTokenRateLimiter] = {} + + +def get_rate_limiter(account: ResolvedSynologyChatAccount) -> UserRateLimiter: + rl = _rate_limiters.get(account.account_id) + if rl is None or rl.max_requests() != account.rate_limit_per_minute: + if rl: + rl.clear() + rl = UserRateLimiter(max_per_window=account.rate_limit_per_minute) + _rate_limiters[account.account_id] = rl + return rl + + +def get_invalid_token_limiter(account_id: str) -> InvalidTokenRateLimiter: + limiter = _invalid_token_limiters.get(account_id) + if limiter is None: + limiter = InvalidTokenRateLimiter() + _invalid_token_limiters[account_id] = limiter + return limiter diff --git a/backend/package/yuxi/channel/extensions/synology_chat/session.py b/backend/package/yuxi/channel/extensions/synology_chat/session.py new file mode 100644 index 00000000..9e6a9e2f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/session.py @@ -0,0 +1,32 @@ +from __future__ import annotations + + +def build_session_key(agent_id: str, account_id: str, user_id: str) -> str: + return f"agent:{agent_id}:synology-chat:{account_id}:direct:{user_id}" + + +CHANNEL_FORMAT_INSTRUCTIONS = """ +### Synology Chat Formatting +Synology Chat supports limited formatting. Use these patterns: + +**Links**: Use to create clickable links. + Example: renders as a clickable link. + Markdown links [label](url) are automatically converted to format. + +**File sharing**: Include a publicly accessible URL to share files or images. + The NAS will download and attach the file (max 32 MB). + +**Interactive buttons**: Bot can send action buttons via attachments. + Supported button styles: green, grey, red, orange, blue, teal. + +**Limitations**: +- No markdown, bold, italic, or code blocks +- No message editing after send +- Keep messages under 2000 characters for best readability + +**Best practices**: +- Use short, clear responses (Synology Chat has a minimal UI) +- Use line breaks to separate sections +- Use numbered or bulleted lists for clarity +- Wrap URLs with for user-friendly links +""" diff --git a/backend/package/yuxi/channel/extensions/synology_chat/status.py b/backend/package/yuxi/channel/extensions/synology_chat/status.py new file mode 100644 index 00000000..375aa635 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/status.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + + +async def probe(account: dict) -> bool: + token = account.get("token", "") + incoming_url = account.get("incoming_url", "") + return bool(token) and bool(incoming_url) + + +def build_summary(account: dict) -> dict: + configured = bool(account.get("token")) and bool(account.get("incoming_url")) + return { + "configured": configured, + "has_token": bool(account.get("token")), + "has_webhook_url": bool(account.get("incoming_url")), + "bot_name": account.get("bot_name", "ForcePilot"), + "dm_policy": account.get("dm_policy", "allowlist"), + "nas_host": account.get("nas_host", ""), + "allow_insecure_ssl": account.get("allow_insecure_ssl", False), + } \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/synology_chat/types.py b/backend/package/yuxi/channel/extensions/synology_chat/types.py new file mode 100644 index 00000000..4ad2af1d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/types.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + +ACCOUNT_ID_DEFAULT = "default" +WEBHOOK_PATH_DEFAULT = "/webhook/synology" +RATE_LIMIT_DEFAULT = 30 +BOT_NAME_DEFAULT = "ForcePilot" +INPUT_MAX_LENGTH = 4000 +PREAUTH_BODY_MAX_BYTES = 64 * 1024 +PREAUTH_BODY_TIMEOUT_S = 5 +AGENT_TIMEOUT_S = 120 +AGENT_SYNC_TIMEOUT_S = 5 +MIN_SEND_INTERVAL_MS = 500 +MAX_RETRIES = 3 +BASE_RETRY_DELAY_S = 0.3 +HTTP_TIMEOUT_S = 30 +CACHE_TTL_S = 300 + +DM_POLICY_OPEN = "open" +DM_POLICY_ALLOWLIST = "allowlist" +DM_POLICY_DISABLED = "disabled" + +WEBHOOK_PATH_SOURCE_DEFAULT = "default" +WEBHOOK_PATH_SOURCE_INHERITED = "inherited-base" +WEBHOOK_PATH_SOURCE_EXPLICIT = "explicit" + + +@dataclass +class SynologyWebhookPayload: + token: str = "" + user_id: str = "" + text: str = "" + username: str = "" + chat_type: str = "direct" + channel_id: str = "" + channel_name: str = "" + post_id: str = "" + timestamp: str = "" + trigger_word: str = "" + + +@dataclass +class ResolvedSynologyChatAccount: + account_id: str + enabled: bool = True + token: str = "" + incoming_url: str = "" + nas_host: str = "localhost" + webhook_path: str = WEBHOOK_PATH_DEFAULT + webhook_path_source: str = WEBHOOK_PATH_SOURCE_DEFAULT + dm_policy: str = DM_POLICY_ALLOWLIST + allowed_user_ids: list[str] = field(default_factory=list) + rate_limit_per_minute: int = RATE_LIMIT_DEFAULT + bot_name: str = BOT_NAME_DEFAULT + allow_insecure_ssl: bool = False + dangerously_allow_name_matching: bool = False + dangerously_allow_inherited_webhook_path: bool = False + + +class WebhookParseError(Exception): + pass diff --git a/backend/package/yuxi/channel/extensions/synology_chat/webhook.py b/backend/package/yuxi/channel/extensions/synology_chat/webhook.py new file mode 100644 index 00000000..835170d4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/synology_chat/webhook.py @@ -0,0 +1,434 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import time +from urllib.parse import parse_qs + +from fastapi import APIRouter, Request, Response +from fastapi.responses import JSONResponse + +from yuxi.channel.extensions.synology_chat.accounts import resolve_account +from yuxi.channel.extensions.synology_chat.client import ( + resolve_legacy_webhook_name_to_chat_user_id, + send_message, +) +from yuxi.channel.extensions.synology_chat.security import ( + authorize_dm, + get_invalid_token_limiter, + get_rate_limiter, + sanitize_input, + validate_token, +) +from yuxi.channel.extensions.synology_chat.types import ( + AGENT_SYNC_TIMEOUT_S, + AGENT_TIMEOUT_S, + PREAUTH_BODY_MAX_BYTES, + PREAUTH_BODY_TIMEOUT_S, + ResolvedSynologyChatAccount, + SynologyWebhookPayload, + WebhookParseError, +) +from yuxi.channel.runtime.manager import gateway +from yuxi.channel.message.models import PeerInfo, PeerKind, UnifiedMessage + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/webhook/synology-chat", tags=["synology-chat"]) + +_MAX_IN_FLIGHT_PER_ACCOUNT = 10 +_in_flight_semaphores: dict[str, asyncio.Semaphore] = {} + + +def _get_in_flight_semaphore(account_id: str) -> asyncio.Semaphore: + sem = _in_flight_semaphores.get(account_id) + if sem is None: + sem = asyncio.Semaphore(_MAX_IN_FLIGHT_PER_ACCOUNT) + _in_flight_semaphores[account_id] = sem + return sem + + +def _extract_field(body_data: dict, query_params, aliases: list[str]) -> str: + for alias in aliases: + val = body_data.get(alias, "") + if val: + return str(val) + for alias in aliases: + val = query_params.get(alias, "") + if val: + return val + return "" + + +def _extract_token(request: Request, body_data: dict) -> str: + token = body_data.get("token", "") + if token: + return str(token) + token = request.query_params.get("token", "") + if token: + return token + for header_name in ["x-synology-token", "x-webhook-token", "x-openclaw-token"]: + token = request.headers.get(header_name, "") + if token: + return token + auth = request.headers.get("authorization", "") + if auth.lower().startswith("bearer "): + return auth[7:] + return "" + + +def _is_action_callback(body_data: dict) -> bool: + return "actions" in body_data and "callback_id" in body_data and "token" in body_data + + +async def _read_body_data(request: Request) -> dict: + content_type = request.headers.get("content-type", "") + + try: + raw_body = await asyncio.wait_for(request.body(), timeout=PREAUTH_BODY_TIMEOUT_S) + except TimeoutError: + raise WebhookParseError("Body read timeout") + + if len(raw_body) > PREAUTH_BODY_MAX_BYTES: + raise WebhookParseError("Body too large") + + body_str = raw_body.decode("utf-8", errors="replace") + + if "application/json" in content_type: + body_data = json.loads(body_str) + elif "application/x-www-form-urlencoded" in content_type: + body_data = {k: v[0] if v else "" for k, v in parse_qs(body_str).items()} + else: + try: + body_data = json.loads(body_str) + except json.JSONDecodeError: + body_data = {k: v[0] if v else "" for k, v in parse_qs(body_str).items()} + + return body_data + + +async def parse_payload(request: Request) -> SynologyWebhookPayload: + body_data = await _read_body_data(request) + + token = _extract_token(request, body_data) + query_params = request.query_params + user_id = _extract_field(body_data, query_params, ["user_id", "userId", "user"]) + text = _extract_field(body_data, query_params, ["text", "message", "content"]) + username = _extract_field(body_data, query_params, ["username", "user_name", "name"]) + + if not token: + raise WebhookParseError("Missing required field: token") + if not user_id: + raise WebhookParseError("Missing required field: user_id") + + chat_type = body_data.get("chat_type", "direct") + if chat_type == "direct" and not text: + raise WebhookParseError("Missing required field: text") + + return SynologyWebhookPayload( + token=token, + user_id=user_id, + text=text, + username=username, + chat_type=chat_type, + channel_id=body_data.get("channel_id", ""), + channel_name=_extract_field(body_data, query_params, ["channel_name", "channelName"]), + post_id=_extract_field(body_data, query_params, ["post_id", "postId"]), + timestamp=_extract_field(body_data, query_params, ["timestamp", "ts"]), + trigger_word=_extract_field(body_data, query_params, ["trigger_word", "triggerWord"]), + ) + + +async def parse_action_callback(request: Request) -> dict: + body_data = await _read_body_data(request) + + token = _extract_token(request, body_data) + if not token or not body_data.get("callback_id"): + raise WebhookParseError("Missing required fields: token, callback_id") + + user = body_data.get("user", {}) + return { + "token": token, + "callback_id": body_data.get("callback_id", ""), + "post_id": body_data.get("post_id", ""), + "actions": body_data.get("actions", []), + "user_id": str(user.get("user_id", "")) if isinstance(user, dict) else "", + "username": user.get("username", "") if isinstance(user, dict) else "", + } + + +async def _resolve_send_user_id( + account: ResolvedSynologyChatAccount, + payload: SynologyWebhookPayload, +) -> str: + if not account.dangerously_allow_name_matching: + return payload.user_id + + resolved = await resolve_legacy_webhook_name_to_chat_user_id( + incoming_url=account.incoming_url, + webhook_username=payload.username, + allow_insecure_ssl=account.allow_insecure_ssl, + ) + return resolved or payload.user_id + + +async def _handle_action_callback( + request: Request, + account_id: str, + body_data: dict, + processor, +) -> Response: + try: + callback = await parse_action_callback(request) + except WebhookParseError as e: + return JSONResponse({"error": str(e)}, status_code=400) + + account = _resolve_account_for_request(account_id) + + if not validate_token(callback["token"], account.token): + return JSONResponse({"error": "Invalid token"}, status_code=401) + + user_id = callback["user_id"] + allowed, reason = authorize_dm(account, user_id) + if not allowed: + return JSONResponse({"error": reason}, status_code=403) + + rl = get_rate_limiter(account) + if not rl.allow(user_id): + return JSONResponse({"error": "Rate limited"}, status_code=429) + + actions = callback.get("actions", []) + action_value = actions[0].get("value", "") if actions else "" + callback_id = callback.get("callback_id", "") + + logger.info( + "Action callback: callback_id=%s user_id=%s action=%s", + callback_id, + user_id, + action_value, + ) + + if processor is not None: + metadata = { + "ChatType": "direct", + "CommandAuthorized": True, + "InteractionType": "button_callback", + "CallbackId": callback_id, + "ActionValue": action_value, + "PostId": callback.get("post_id", ""), + "user_id": user_id, + "username": callback.get("username", ""), + } + asyncio.create_task( + _dispatch_to_agent( + processor, + account, + None, + f"[callback:{callback_id}]:{action_value}", + metadata=metadata, + ), + name=f"synology-agent-cb-{account.account_id}-{user_id}", + ) + + return Response(status_code=204) + + +def _build_callback_update_response( + callback_id: str, + text: str, + attachments: list[dict] | None = None, +) -> dict: + payload: dict = {"text": text} + if attachments: + payload["attachments"] = attachments + return payload + + +@router.post("/{account_id}") +async def synology_chat_webhook( + request: Request, + account_id: str = "default", +) -> Response: + processor = gateway._processor + + sem = _get_in_flight_semaphore(account_id) + if sem.locked(): + return JSONResponse({"error": "Too many concurrent requests"}, status_code=503) + + async with sem: + try: + body_data = await _read_body_data(request) + except WebhookParseError as e: + return JSONResponse({"error": str(e)}, status_code=400) + except Exception: + logger.exception("Failed to read webhook body") + return JSONResponse({"error": "Bad request"}, status_code=400) + + if _is_action_callback(body_data): + return await _handle_action_callback(request, account_id, body_data, processor) + + try: + payload = await parse_payload(request) + except WebhookParseError as e: + return JSONResponse({"error": str(e)}, status_code=400) + except Exception: + logger.exception("Failed to parse webhook payload") + return JSONResponse({"error": "Bad request"}, status_code=400) + + account = _resolve_account_for_request(account_id) + + if not validate_token(payload.token, account.token): + limiter = get_invalid_token_limiter(account.account_id) + client_ip = request.client.host if request.client else "unknown" + if not limiter.allow(client_ip): + return JSONResponse({"error": "Too many invalid token attempts"}, status_code=429) + return JSONResponse({"error": "Invalid token"}, status_code=401) + + allowed, reason = authorize_dm(account, payload.user_id) + if not allowed: + return JSONResponse({"error": reason}, status_code=403) + + rl = get_rate_limiter(account) + if not rl.allow(payload.user_id): + return JSONResponse({"error": "Rate limited"}, status_code=429) + + sanitized_text = sanitize_input(payload.text) + + if processor is not None: + sync_reply = await _dispatch_to_agent_sync(processor, account, payload, sanitized_text) + if sync_reply is not None: + return JSONResponse(sync_reply, status_code=200) + + asyncio.create_task( + _dispatch_to_agent(processor, account, payload, sanitized_text), + name=f"synology-agent-{account.account_id}-{payload.user_id}", + ) + else: + logger.warning("Message processor not available, cannot dispatch") + + return Response(status_code=204) + + +async def _dispatch_to_agent( + processor, + account: ResolvedSynologyChatAccount, + payload: SynologyWebhookPayload | None, + sanitized_text: str, + metadata: dict | None = None, +) -> None: + if payload is not None: + send_user_id = await _resolve_send_user_id(account, payload) + msg_metadata: dict = { + "ChatType": payload.chat_type, + "CommandAuthorized": True, + "webhook_user_id": payload.user_id, + } + if payload.channel_id: + msg_metadata["ChannelId"] = payload.channel_id + if payload.channel_name: + msg_metadata["ChannelName"] = payload.channel_name + display_name = payload.username + user_id = payload.user_id + else: + send_user_id = (metadata or {}).get("user_id", "") + msg_metadata = metadata or {} + display_name = msg_metadata.get("username", "") + user_id = send_user_id + + try: + msg = UnifiedMessage( + msg_id=f"sc-{int(time.time() * 1000)}", + channel_type="synology-chat", + account_id=account.account_id, + content=sanitized_text, + sender=PeerInfo(id=user_id, kind=PeerKind.DIRECT, display_name=display_name), + body_for_agent=sanitized_text, + metadata=msg_metadata, + ) + + await asyncio.wait_for(processor.process(msg), timeout=AGENT_TIMEOUT_S) + + except TimeoutError: + logger.error( + "Agent response timeout (120s) for %s/%s", + account.account_id, + user_id, + ) + if account.incoming_url: + await send_message( + incoming_url=account.incoming_url, + text="Request timed out. Please try again.", + user_id=send_user_id, + allow_insecure_ssl=account.allow_insecure_ssl, + ) + except Exception: + logger.exception( + "Failed to process message for %s/%s", + account.account_id, + user_id, + ) + if account.incoming_url: + await send_message( + incoming_url=account.incoming_url, + text="Sorry, an error occurred while processing your message.", + user_id=send_user_id, + allow_insecure_ssl=account.allow_insecure_ssl, + ) + + +async def _dispatch_to_agent_sync( + processor, + account: ResolvedSynologyChatAccount, + payload: SynologyWebhookPayload, + sanitized_text: str, +) -> dict | None: + msg = UnifiedMessage( + msg_id=f"sc-{payload.timestamp or int(time.time() * 1000)}", + channel_type="synology-chat", + account_id=account.account_id, + content=sanitized_text, + sender=PeerInfo( + id=payload.user_id, + kind=PeerKind.DIRECT, + display_name=payload.username, + ), + body_for_agent=sanitized_text, + metadata={ + "ChatType": payload.chat_type, + "CommandAuthorized": True, + "webhook_user_id": payload.user_id, + "SyncResponse": True, + }, + ) + + try: + await asyncio.wait_for(processor.process(msg), timeout=AGENT_SYNC_TIMEOUT_S) + except TimeoutError: + logger.info("Sync response timeout for %s/%s, falling back to async", account.account_id, payload.user_id) + return None + except Exception: + logger.exception("Sync dispatch failed for %s/%s", account.account_id, payload.user_id) + return None + + return None + + +_ACCOUNT_CACHE: dict[str, ResolvedSynologyChatAccount] = {} +_ACCOUNT_CACHE_CONFIG_HASH: str = "" + + +def _resolve_account_for_request(account_id: str) -> ResolvedSynologyChatAccount: + global _ACCOUNT_CACHE, _ACCOUNT_CACHE_CONFIG_HASH + + if account_id in _ACCOUNT_CACHE: + return _ACCOUNT_CACHE[account_id] + + account = resolve_account(gateway.global_config, account_id) + _ACCOUNT_CACHE[account_id] = account + return account + + +def invalidate_account_cache() -> None: + global _ACCOUNT_CACHE + _ACCOUNT_CACHE.clear()