From 6fc9a4a94c66eb9cd89fac2ee0c8b6daf2b68816 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 11:29:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E6=B7=BB=E5=8A=A0=20Nextcloud?= =?UTF-8?q?=20Talk=20=E6=B8=A0=E9=81=93=E6=89=A9=E5=B1=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 Nextcloud Talk 渠道扩展，支持在 Yuxi 平台中集成 Nextcloud Talk 即时通讯渠道。 包含以下功能模块： - accounts: 账户管理 - bot_admin: Bot 管理 - chat_api: 聊天 API 封装 - config_schema: 配置模式 - conversation: 会话管理 - gateway: SSE/WebSocket 网关接入 - inbound: 入站消息处理 - send: 消息发送 - webhook_server: Webhook 服务 - format: 消息格式转换 - normalize: 消息规范化 - poll_api: 投票 API - reaction_api: 表情反应 API - policy: 安全策略 - probe: 健康探测 - replay_guard: 重放防护 - signature: 签名验证 - room_info: 房间信息 - types: 类型定义 --- .../extensions/nextcloud_talk/__init__.py | 545 ++++++++++++++++++ .../extensions/nextcloud_talk/accounts.py | 75 +++ .../extensions/nextcloud_talk/bot_admin.py | 120 ++++ .../extensions/nextcloud_talk/chat_api.py | 169 ++++++ .../nextcloud_talk/config_schema.py | 64 ++ .../extensions/nextcloud_talk/conversation.py | 290 ++++++++++ .../extensions/nextcloud_talk/format.py | 62 ++ .../extensions/nextcloud_talk/gateway.py | 208 +++++++ .../extensions/nextcloud_talk/inbound.py | 159 +++++ .../extensions/nextcloud_talk/normalize.py | 28 + .../extensions/nextcloud_talk/plugin.json | 23 + .../extensions/nextcloud_talk/policy.py | 54 ++ .../extensions/nextcloud_talk/poll_api.py | 120 ++++ .../extensions/nextcloud_talk/probe.py | 154 +++++ .../extensions/nextcloud_talk/reaction_api.py | 100 ++++ .../extensions/nextcloud_talk/replay_guard.py | 104 ++++ .../extensions/nextcloud_talk/room_info.py | 97 ++++ .../channel/extensions/nextcloud_talk/send.py | 155 +++++ .../extensions/nextcloud_talk/signature.py | 49 ++ .../extensions/nextcloud_talk/types.py | 60 ++ .../nextcloud_talk/webhook_server.py | 251 ++++++++ 21 files changed, 2887 insertions(+) create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/__init__.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/accounts.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/bot_admin.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/chat_api.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/config_schema.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/conversation.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/format.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/gateway.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/inbound.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/normalize.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/plugin.json create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/policy.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/poll_api.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/probe.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/reaction_api.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/replay_guard.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/room_info.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/send.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/signature.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/types.py create mode 100644 backend/package/yuxi/channel/extensions/nextcloud_talk/webhook_server.py diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/__init__.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/__init__.py new file mode 100644 index 00000000..59ea1b48 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/__init__.py @@ -0,0 +1,545 @@ +from __future__ import annotations + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.context import ChannelContext +from yuxi.channel.errors import classify_error +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.nextcloud_talk.accounts import resolve_nextcloud_talk_account +from yuxi.channel.extensions.nextcloud_talk.bot_admin import ( + disable_bot_in_room, + enable_bot_in_room, + list_room_bots, + list_server_bots, +) +from yuxi.channel.extensions.nextcloud_talk.chat_api import ( + delete_chat_message, + edit_chat_message, + fetch_chat_history, + fetch_message_context, + mark_chat_read, +) +from yuxi.channel.extensions.nextcloud_talk.config_schema import NextcloudTalkConfig +from yuxi.channel.extensions.nextcloud_talk.conversation import ( + create_conversation, + delete_conversation, + list_conversations, + rename_conversation, +) +from yuxi.channel.extensions.nextcloud_talk.gateway import NextcloudTalkGateway +from yuxi.channel.extensions.nextcloud_talk.normalize import strip_nextcloud_talk_target_prefix +from yuxi.channel.extensions.nextcloud_talk.poll_api import close_poll, create_poll, vote_poll +from yuxi.channel.extensions.nextcloud_talk.probe import probe_nextcloud_talk +from yuxi.channel.extensions.nextcloud_talk.reaction_api import list_reactions +from yuxi.channel.extensions.nextcloud_talk.send import ( + NextcloudTalkSendError, + send_message_nextcloud_talk, + send_reaction_nextcloud_talk, +) +from yuxi.channel.message.models import PeerKind +from yuxi.channel.plugins.registry import ChannelPluginRegistry +from yuxi.channel.protocols import ClassifiedError, ErrorSeverity, SessionResolution + + +class NextcloudTalkPlugin(BaseChannelPlugin): + id = "nextcloud-talk" + name = "Nextcloud Talk" + order = 65 + label = "Nextcloud Talk (Webhook Bot)" + aliases = ["nc-talk", "nc"] + resolve_reply_to_mode = "native" + + def __init__(self): + self._gateway = NextcloudTalkGateway() + self._config: dict = {} + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct", "group"], + message_types=["text"], + reactions=True, + threads=False, + media=True, + native_commands=False, + block_streaming=True, + ) + + def _get_gateway(self) -> NextcloudTalkGateway: + gateway = self._gateway + if gateway.account is None: + raise RuntimeError("Nextcloud Talk gateway not started") + return gateway + + def _require_api_gateway(self) -> NextcloudTalkGateway: + gateway = self._get_gateway() + if not gateway.account.api_user or not gateway.account.api_password: + raise RuntimeError("Nextcloud Talk API credentials not configured") + return gateway + + # ── ConfigProtocol ──────────────────────────────────── + + def list_account_ids(self, config: dict) -> list[str]: + return ["default"] + + async def resolve_account(self, account_id: str) -> dict: + channel_cfg = self._config.get("channels", {}).get("nextcloud-talk", {}) + cfg = NextcloudTalkConfig(**channel_cfg) + account = resolve_nextcloud_talk_account(cfg, account_id) + return { + "account_id": account.account_id, + "enabled": account.enabled, + "name": account.name, + "configured": account.configured, + "base_url": account.base_url, + "bot_secret": account.bot_secret, + "secret_source": account.secret_source, + "api_user": account.api_user, + "api_password": account.api_password, + "api_password_source": account.api_password_source, + } + + def is_configured(self, account: dict) -> bool: + return bool(account.get("base_url") and account.get("bot_secret")) + + def is_enabled(self, account: dict) -> bool: + return account.get("enabled", True) + + def disabled_reason(self, account: dict) -> str: + if not account.get("base_url"): + return "base_url is required" + if not account.get("bot_secret"): + return "bot_secret is required" + return "" + + async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None: + channel_cfg = config.get("channels", {}).get("nextcloud-talk", {}) + cfg = NextcloudTalkConfig(**channel_cfg) + return cfg.allow_from if cfg.allow_from else None + + def describe_account(self, account: dict) -> dict: + return { + "account_id": account.get("account_id", "default"), + "base_url": account.get("base_url", ""), + "secret_source": account.get("secret_source", "none"), + } + + # ── ConfigSchemaProtocol ────────────────────────────── + + def config_schema(self) -> dict: + return NextcloudTalkConfig.model_json_schema() + + # ── ErrorHandlingProtocol ───────────────────────────── + + def classify_error(self, error: BaseException) -> ClassifiedError: + if isinstance(error, NextcloudTalkSendError): + status = error.status_code + if status == 429: + severity = ErrorSeverity.RATE_LIMITED + elif status in (401, 403): + severity = ErrorSeverity.FORBIDDEN + elif status in (400, 404): + severity = ErrorSeverity.FATAL + else: + severity = ErrorSeverity.RETRYABLE + return ClassifiedError(severity=severity, original_error=error, error_message=str(error)) + if isinstance(error, RuntimeError): + return ClassifiedError(severity=ErrorSeverity.FATAL, original_error=error, error_message=str(error)) + return classify_error(error) + + # ── GatewayProtocol ─────────────────────────────────── + + async def start(self, ctx: ChannelContext) -> object: + return await self._gateway.start(ctx) + + async def stop(self, ctx: ChannelContext) -> None: + await self._gateway.stop(ctx) + + # ── 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, + ) -> None: + gateway = self._get_gateway() + await send_message_nextcloud_talk( + to=target_id, + text=content, + base_url=gateway.account.base_url, + secret=gateway.account.bot_secret, + reply_to=reply_to_id, + dangerously_allow_private_network=(gateway.cfg.dangerously_allow_private_network if gateway.cfg else False), + ) + + 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: + text = f"\n\nAttachment: <{media_url}>" + await self.send_text(target_id, text, reply_to_id=reply_to_id) + + async def send_file( + self, + target_id: str, + file_path: str, + file_name: str | None = None, + reply_to_id: str | None = None, + account_id: str | None = None, + ) -> None: + raise NotImplementedError( + "Nextcloud Talk Bot API does not support direct file upload. Use send_media() to share a URL link instead." + ) + + async def send_reaction( + self, + target_id: str, + message_id: str, + reaction: str, + *, + account_id: str | None = None, + ) -> bool: + gateway = self._get_gateway() + return await send_reaction_nextcloud_talk( + room_token=strip_nextcloud_talk_target_prefix(target_id), + message_id=message_id, + reaction=reaction, + base_url=gateway.account.base_url, + secret=gateway.account.bot_secret, + dangerously_allow_private_network=(gateway.cfg.dangerously_allow_private_network if gateway.cfg else False), + ) + + # ── Chat API ────────────────────────────────────────── + + async def fetch_history( + self, + room_token: str, + *, + limit: int = 100, + account_id: str | None = None, + ) -> list[dict]: + gateway = self._require_api_gateway() + return await fetch_chat_history( + room_token=strip_nextcloud_talk_target_prefix(room_token), + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + limit=limit, + ) + + async def fetch_message_context( + self, + room_token: str, + message_id: str, + *, + limit: int = 25, + account_id: str | None = None, + ) -> list[dict]: + gateway = self._require_api_gateway() + return await fetch_message_context( + room_token=strip_nextcloud_talk_target_prefix(room_token), + message_id=message_id, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + limit=limit, + ) + + async def delete_message( + self, + room_token: str, + message_id: str, + *, + account_id: str | None = None, + ) -> bool: + gateway = self._require_api_gateway() + return await delete_chat_message( + room_token=strip_nextcloud_talk_target_prefix(room_token), + message_id=message_id, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + async def edit_message( + self, + room_token: str, + message_id: str, + text: str, + *, + account_id: str | None = None, + ) -> bool: + gateway = self._require_api_gateway() + return await edit_chat_message( + room_token=strip_nextcloud_talk_target_prefix(room_token), + message_id=message_id, + text=text, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + async def mark_read( + self, + room_token: str, + *, + message_id: str | None = None, + account_id: str | None = None, + ) -> bool: + gateway = self._require_api_gateway() + return await mark_chat_read( + room_token=strip_nextcloud_talk_target_prefix(room_token), + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + message_id=message_id, + ) + + # ── Conversation API ────────────────────────────────── + + async def list_conversations(self, *, account_id: str | None = None) -> list[dict]: + gateway = self._require_api_gateway() + return await list_conversations( + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + async def create_conversation( + self, + *, + room_type: int, + invite: str | None = None, + room_name: str | None = None, + account_id: str | None = None, + ) -> dict | None: + gateway = self._require_api_gateway() + return await create_conversation( + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + room_type=room_type, + invite=invite, + room_name=room_name, + ) + + async def delete_conversation( + self, + room_token: str, + *, + account_id: str | None = None, + ) -> bool: + gateway = self._require_api_gateway() + return await delete_conversation( + room_token=strip_nextcloud_talk_target_prefix(room_token), + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + async def rename_conversation( + self, + room_token: str, + room_name: str, + *, + account_id: str | None = None, + ) -> bool: + gateway = self._require_api_gateway() + return await rename_conversation( + room_token=strip_nextcloud_talk_target_prefix(room_token), + room_name=room_name, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + # ── Reaction API ────────────────────────────────────── + + async def list_reactions( + self, + room_token: str, + message_id: str, + *, + account_id: str | None = None, + ) -> dict: + gateway = self._require_api_gateway() + return await list_reactions( + room_token=strip_nextcloud_talk_target_prefix(room_token), + message_id=message_id, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + # ── Poll API ────────────────────────────────────────── + + async def create_poll( + self, + room_token: str, + question: str, + options: list[str], + *, + result_mode: int = 0, + max_votes: int = 1, + account_id: str | None = None, + ) -> dict | None: + gateway = self._require_api_gateway() + return await create_poll( + room_token=strip_nextcloud_talk_target_prefix(room_token), + question=question, + options=options, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + result_mode=result_mode, + max_votes=max_votes, + ) + + async def vote_poll( + self, + room_token: str, + poll_id: str, + option_ids: list[int], + *, + account_id: str | None = None, + ) -> dict | None: + gateway = self._require_api_gateway() + return await vote_poll( + room_token=strip_nextcloud_talk_target_prefix(room_token), + poll_id=poll_id, + option_ids=option_ids, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + async def close_poll( + self, + room_token: str, + poll_id: str, + *, + account_id: str | None = None, + ) -> dict | None: + gateway = self._require_api_gateway() + return await close_poll( + room_token=strip_nextcloud_talk_target_prefix(room_token), + poll_id=poll_id, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + # ── Bot Admin API ───────────────────────────────────── + + async def list_server_bots(self, *, account_id: str | None = None) -> list[dict]: + gateway = self._require_api_gateway() + return await list_server_bots( + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + async def list_room_bots( + self, + room_token: str, + *, + account_id: str | None = None, + ) -> list[dict]: + gateway = self._require_api_gateway() + return await list_room_bots( + room_token=strip_nextcloud_talk_target_prefix(room_token), + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + async def enable_bot( + self, + room_token: str, + bot_id: int, + *, + account_id: str | None = None, + ) -> bool: + gateway = self._require_api_gateway() + return await enable_bot_in_room( + room_token=strip_nextcloud_talk_target_prefix(room_token), + bot_id=bot_id, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + async def disable_bot( + self, + room_token: str, + bot_id: int, + *, + account_id: str | None = None, + ) -> bool: + gateway = self._require_api_gateway() + return await disable_bot_in_room( + room_token=strip_nextcloud_talk_target_prefix(room_token), + bot_id=bot_id, + base_url=gateway.account.base_url, + api_user=gateway.account.api_user, + api_password=gateway.account.api_password, + ) + + # ── StatusProtocol ──────────────────────────────────── + + async def probe(self, account: dict) -> bool: + probe_result = await probe_nextcloud_talk(account) + return probe_result.ok + + def build_summary(self, snapshot: object) -> dict: + return { + "channel": "nextcloud-talk", + "configured": self._gateway.account is not None, + } + + async def check_ready(self, account_id: str) -> bool: + return self._gateway.check_health() + + # ── AgentPromptProtocol ─────────────────────────────── + + @property + def channel_format_instructions(self) -> str | None: + return ( + "You are responding via Nextcloud Talk. Messages support basic Markdown. " + "Be concise. The bot name is ForcePilot." + ) + + # ── LifecycleProtocol ───────────────────────────────── + + @property + def config_prefixes(self) -> list[str]: + return ["channels.nextcloud-talk"] + + async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None: + self._config = next_cfg + + # ── MessagingProtocol ───────────────────────────────── + + def resolve_session(self, msg): + if hasattr(msg, "sender") and hasattr(msg.sender, "kind"): + kind = msg.sender.kind + sender_id = msg.sender.id if hasattr(msg.sender, "id") else "" + sender_name = msg.sender.display_name if hasattr(msg.sender, "display_name") else "" + if kind == PeerKind.DIRECT: + return SessionResolution(kind="direct", conversation_id=sender_id, label=sender_name) + + gid = "" + if hasattr(msg, "group") and msg.group: + gid = msg.group.id or "unknown" + return SessionResolution(kind="group", conversation_id=gid or "unknown") + + +nextcloud_talk_plugin = ChannelPluginRegistry.register(NextcloudTalkPlugin()) diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/accounts.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/accounts.py new file mode 100644 index 00000000..0fb0febd --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/accounts.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import logging +import os +from pathlib import Path + +from yuxi.channel.extensions.nextcloud_talk.config_schema import NextcloudTalkConfig +from yuxi.channel.extensions.nextcloud_talk.types import ResolvedNextcloudTalkAccount + +logger = logging.getLogger(__name__) + + +def _read_secret_file(file_path: str) -> str: + p = Path(file_path) + if not p.is_file(): + logger.warning("Nextcloud Talk: secret file not found: %s", file_path) + return "" + if p.is_symlink(): + logger.warning("Nextcloud Talk: secret file is a symlink, refusing to read: %s", file_path) + return "" + try: + return p.read_text(encoding="utf-8").strip() + except Exception as e: + logger.warning("Nextcloud Talk: failed to read secret file %s: %s", file_path, e) + return "" + + +def _resolve_bot_secret(cfg: NextcloudTalkConfig, account_id: str) -> tuple[str, str]: + if account_id == "default": + env_val = os.environ.get("NEXTCLOUD_TALK_BOT_SECRET") + if env_val: + return env_val, "env" + if cfg.bot_secret_file: + content = _read_secret_file(cfg.bot_secret_file) + if content: + return content, "file" + if cfg.bot_secret: + return cfg.bot_secret, "config" + return "", "none" + + +def _resolve_api_password(cfg: NextcloudTalkConfig) -> tuple[str, str]: + if cfg.api_password: + return cfg.api_password, "config" + if cfg.api_password_file: + content = _read_secret_file(cfg.api_password_file) + if content: + return content, "file" + return "", "none" + + +def resolve_nextcloud_talk_account( + cfg: NextcloudTalkConfig, account_id: str = "default" +) -> ResolvedNextcloudTalkAccount: + bot_secret, secret_source = _resolve_bot_secret(cfg, account_id) + api_password, api_password_source = _resolve_api_password(cfg) + + configured = bool(cfg.base_url and bot_secret) + + return ResolvedNextcloudTalkAccount( + account_id=account_id, + enabled=cfg.enabled, + name=cfg.name or account_id, + configured=configured, + base_url=cfg.base_url, + bot_secret=bot_secret, + secret_source=secret_source, + api_user=cfg.api_user, + api_password=api_password, + api_password_source=api_password_source, + ) + + +def list_nextcloud_talk_account_ids(cfg: NextcloudTalkConfig) -> list[str]: + return ["default"] diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/bot_admin.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/bot_admin.py new file mode 100644 index 00000000..135e018a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/bot_admin.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import base64 +import logging +from urllib.parse import urljoin + +import httpx + +logger = logging.getLogger(__name__) + +OCS_BOT_ADMIN_PATH = "/ocs/v2.php/apps/spreed/api/v1/bot/admin" +OCS_BOT_ROOM_PATH = "/ocs/v2.php/apps/spreed/api/v1/bot/{token}" +OCS_BOT_MANAGE_PATH = "/ocs/v2.php/apps/spreed/api/v1/bot/{token}/{bot_id}" + +DEFAULT_TIMEOUT = 30.0 + + +def _auth_headers(api_user: str, api_password: str) -> dict: + auth = base64.b64encode(f"{api_user}:{api_password}".encode()).decode() + return { + "Authorization": f"Basic {auth}", + "OCS-APIRequest": "true", + "Accept": "application/json", + } + + +async def list_server_bots( + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> list[dict]: + url = urljoin(base_url.rstrip("/") + "/", OCS_BOT_ADMIN_PATH) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.get( + url, + headers=_auth_headers(api_user, api_password), + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk list server bots failed: HTTP %s", resp.status_code) + return [] + + data = resp.json() + return data.get("ocs", {}).get("data", []) + + +async def list_room_bots( + room_token: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> list[dict]: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_BOT_ROOM_PATH.format(token=room_token), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.get( + url, + headers=_auth_headers(api_user, api_password), + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk list room bots failed: HTTP %s", resp.status_code) + return [] + + data = resp.json() + return data.get("ocs", {}).get("data", []) + + +async def enable_bot_in_room( + room_token: str, + bot_id: int, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_BOT_MANAGE_PATH.format(token=room_token, bot_id=bot_id), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def disable_bot_in_room( + room_token: str, + bot_id: int, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_BOT_MANAGE_PATH.format(token=room_token, bot_id=bot_id), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.delete( + url, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/chat_api.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/chat_api.py new file mode 100644 index 00000000..54d5fdad --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/chat_api.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import base64 +import logging +from urllib.parse import urljoin + +import httpx + +logger = logging.getLogger(__name__) + +OCS_CHAT_PATH = "/ocs/v2.php/apps/spreed/api/v1/chat/{token}" +OCS_CHAT_CONTEXT_PATH = "/ocs/v2.php/apps/spreed/api/v1/chat/{token}/{message_id}/context" +OCS_CHAT_SHARE_PATH = "/ocs/v2.php/apps/spreed/api/v1/chat/{token}/share" +OCS_CHAT_READ_PATH = "/ocs/v2.php/apps/spreed/api/v1/chat/{token}/read" + +DEFAULT_TIMEOUT = 30.0 + + +def _basic_auth(api_user: str, api_password: str) -> str: + return base64.b64encode(f"{api_user}:{api_password}".encode()).decode() + + +def _auth_headers(api_user: str, api_password: str) -> dict: + return { + "Authorization": f"Basic {_basic_auth(api_user, api_password)}", + "OCS-APIRequest": "true", + "Accept": "application/json", + } + + +async def fetch_chat_history( + room_token: str, + base_url: str, + api_user: str, + api_password: str, + *, + limit: int = 100, + look_into_future: int = 0, + timeout: float = DEFAULT_TIMEOUT, +) -> list[dict]: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_CHAT_PATH.format(token=room_token), + ) + params = { + "limit": limit, + "lookIntoFuture": look_into_future, + "includeLastKnown": "1", + } + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.get( + url, + headers=_auth_headers(api_user, api_password), + params=params, + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk chat history fetch failed: HTTP %s", resp.status_code) + return [] + + data = resp.json() + return data.get("ocs", {}).get("data", []) + + +async def fetch_message_context( + room_token: str, + message_id: str, + base_url: str, + api_user: str, + api_password: str, + *, + limit: int = 25, + timeout: float = DEFAULT_TIMEOUT, +) -> list[dict]: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_CHAT_CONTEXT_PATH.format(token=room_token, message_id=message_id), + ) + params = {"limit": limit} + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.get( + url, + headers=_auth_headers(api_user, api_password), + params=params, + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk message context fetch failed: HTTP %s", resp.status_code) + return [] + + data = resp.json() + return data.get("ocs", {}).get("data", []) + + +async def delete_chat_message( + room_token: str, + message_id: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + f"/ocs/v2.php/apps/spreed/api/v1/chat/{room_token}/{message_id}", + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.delete( + url, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def edit_chat_message( + room_token: str, + message_id: str, + text: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + f"/ocs/v2.php/apps/spreed/api/v1/chat/{room_token}/{message_id}", + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.put( + url, + json={"message": text}, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def mark_chat_read( + room_token: str, + base_url: str, + api_user: str, + api_password: str, + *, + message_id: str | None = None, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_CHAT_READ_PATH.format(token=room_token), + ) + body = {} + if message_id: + body["lastReadMessage"] = message_id + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + json=body if body else None, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/config_schema.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/config_schema.py new file mode 100644 index 00000000..b281e884 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/config_schema.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +from enum import StrEnum + +from pydantic import BaseModel, Field + + +class DmPolicy(StrEnum): + PAIRING = "pairing" + OPEN = "open" + ALLOWLIST = "allowlist" + DISABLED = "disabled" + + +class GroupPolicy(StrEnum): + OPEN = "open" + ALLOWLIST = "allowlist" + DISABLED = "disabled" + + +class ChunkMode(StrEnum): + LENGTH = "length" + NEWLINE = "newline" + + +class NextcloudTalkRoomConfig(BaseModel): + require_mention: bool = True + enabled: bool = True + allow_from: list[str] = Field(default_factory=list) + system_prompt: str | None = None + skills: list[str] = Field(default_factory=list) + tools_allow: list[str] = Field(default_factory=list) + tools_deny: list[str] = Field(default_factory=list) + + +class NextcloudTalkConfig(BaseModel): + enabled: bool = True + name: str | None = None + base_url: str = Field(default="", description="Nextcloud 实例 URL") + bot_secret: str | None = Field(None, description="Bot 共享密钥") + bot_secret_file: str | None = Field(None, description="Bot 密钥文件路径") + api_user: str | None = Field(None, description="API 用户名（Room 查询）") + api_password: str | None = Field(None, description="API 密码（Room 查询）") + api_password_file: str | None = Field(None) + + dm_policy: DmPolicy = DmPolicy.PAIRING + allow_from: list[str] = Field(default_factory=list) + group_policy: GroupPolicy = GroupPolicy.ALLOWLIST + group_allow_from: list[str] = Field(default_factory=list) + + webhook_port: int = 8788 + webhook_host: str = "0.0.0.0" + webhook_path: str = "/nextcloud-talk-webhook" + webhook_public_url: str | None = Field(None, description="反向代理后的公开 URL") + + rooms: dict[str, NextcloudTalkRoomConfig] = Field(default_factory=dict) + + text_chunk_limit: int = 4000 + chunk_mode: ChunkMode = ChunkMode.LENGTH + block_streaming: bool = False + + auto_join_message: str | None = Field(None, description="Bot 被加入会话时自动发送的欢迎消息") + + dangerously_allow_private_network: bool = False diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/conversation.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/conversation.py new file mode 100644 index 00000000..183aabeb --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/conversation.py @@ -0,0 +1,290 @@ +from __future__ import annotations + +import base64 +import logging +from urllib.parse import urljoin + +import httpx + +logger = logging.getLogger(__name__) + +OCS_ROOM_PATH = "/ocs/v2.php/apps/spreed/api/v4/room" +OCS_ROOM_TOKEN_PATH = "/ocs/v2.php/apps/spreed/api/v4/room/{token}" + +DEFAULT_TIMEOUT = 30.0 + + +def _auth_headers(api_user: str, api_password: str) -> dict: + auth = base64.b64encode(f"{api_user}:{api_password}".encode()).decode() + return { + "Authorization": f"Basic {auth}", + "OCS-APIRequest": "true", + "Accept": "application/json", + } + + +async def list_conversations( + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> list[dict]: + url = urljoin(base_url.rstrip("/") + "/", OCS_ROOM_PATH) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.get( + url, + headers=_auth_headers(api_user, api_password), + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk list conversations failed: HTTP %s", resp.status_code) + return [] + + data = resp.json() + return data.get("ocs", {}).get("data", []) + + +async def create_conversation( + base_url: str, + api_user: str, + api_password: str, + *, + room_type: int, + invite: str | None = None, + room_name: str | None = None, + timeout: float = DEFAULT_TIMEOUT, +) -> dict | None: + url = urljoin(base_url.rstrip("/") + "/", OCS_ROOM_PATH) + body: dict = {"roomType": room_type} + if invite: + body["invite"] = invite + if room_name: + body["roomName"] = room_name + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + json=body, + headers=_auth_headers(api_user, api_password), + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk create conversation failed: HTTP %s", resp.status_code) + return None + + data = resp.json() + return data.get("ocs", {}).get("data") + + +async def delete_conversation( + room_token: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_ROOM_TOKEN_PATH.format(token=room_token), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.delete( + url, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def rename_conversation( + room_token: str, + room_name: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_ROOM_TOKEN_PATH.format(token=room_token), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.put( + url, + json={"roomName": room_name}, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def set_conversation_description( + room_token: str, + description: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + f"/ocs/v2.php/apps/spreed/api/v4/room/{room_token}/description", + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.put( + url, + json={"description": description}, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def set_conversation_password( + room_token: str, + password: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + f"/ocs/v2.php/apps/spreed/api/v4/room/{room_token}/password", + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.put( + url, + json={"password": password}, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def remove_conversation_password( + room_token: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + f"/ocs/v2.php/apps/spreed/api/v4/room/{room_token}/password", + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.delete( + url, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def set_read_only( + room_token: str, + state: int, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + f"/ocs/v2.php/apps/spreed/api/v4/room/{room_token}/read-only", + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.put( + url, + json={"state": state}, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def set_message_expiration( + room_token: str, + seconds: int, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + f"/ocs/v2.php/apps/spreed/api/v4/room/{room_token}/message-expiration", + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + json={"seconds": seconds}, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def set_favorite( + room_token: str, + base_url: str, + api_user: str, + api_password: str, + *, + favorite: bool = True, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + action = "favorite" if favorite else "unfavorite" + url = urljoin( + base_url.rstrip("/") + "/", + f"/ocs/v2.php/apps/spreed/api/v4/room/{room_token}/{action}", + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def set_notification_level( + room_token: str, + level: int, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + f"/ocs/v2.php/apps/spreed/api/v4/room/{room_token}/notify", + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + json={"level": level}, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/format.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/format.py new file mode 100644 index 00000000..f381a7d2 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/format.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import re + +_SEPARATOR_ROW_RE = re.compile(r"^\|[\s\-:|]+\|$") + + +def _is_separator_row(line: str) -> bool: + stripped = line.strip() + return bool(_SEPARATOR_ROW_RE.match(stripped)) + + +def _parse_table_row(line: str) -> list[str]: + return [cell.strip() for cell in line.strip("|").split("|")] + + +def _table_to_ascii(table_lines: list[str]) -> str: + rows: list[list[str]] = [] + for line in table_lines: + if _is_separator_row(line): + continue + rows.append(_parse_table_row(line)) + + if not rows: + return "" + + col_count = max(len(row) for row in rows) + col_widths = [0] * col_count + for row in rows: + for col_idx in range(col_count): + if col_idx < len(row): + col_widths[col_idx] = max(col_widths[col_idx], len(row[col_idx])) + + result: list[str] = [] + for row in rows: + padded = [row[col_idx].ljust(col_widths[col_idx]) if col_idx < len(row) else "" for col_idx in range(col_count)] + result.append("| " + " | ".join(padded) + " |") + return "\n".join(result) + + +def convert_markdown_tables_to_ascii(text: str) -> str: + lines = text.split("\n") + if not any(line.strip().startswith("|") and line.strip().endswith("|") for line in lines): + return text + + output: list[str] = [] + i = 0 + while i < len(lines): + stripped = lines[i].strip() + if stripped.startswith("|") and stripped.endswith("|"): + table_lines: list[str] = [] + while i < len(lines) and lines[i].strip().startswith("|") and lines[i].strip().endswith("|"): + table_lines.append(lines[i]) + i += 1 + ascii_table = _table_to_ascii(table_lines) + if ascii_table: + output.append(ascii_table) + else: + output.append(lines[i]) + i += 1 + + return "\n".join(output) diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/gateway.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/gateway.py new file mode 100644 index 00000000..071e5d2d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/gateway.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +import logging +from datetime import datetime, UTC + +from aiohttp import web + +from yuxi.channel.context import ChannelContext +from yuxi.channel.extensions.nextcloud_talk.accounts import resolve_nextcloud_talk_account +from yuxi.channel.extensions.nextcloud_talk.config_schema import NextcloudTalkConfig +from yuxi.channel.extensions.nextcloud_talk.inbound import handle_nextcloud_talk_inbound +from yuxi.channel.extensions.nextcloud_talk.replay_guard import ReplayGuard +from yuxi.channel.extensions.nextcloud_talk.room_info import RoomInfoResolver +from yuxi.channel.extensions.nextcloud_talk.send import ( + NextcloudTalkSendError, + send_message_nextcloud_talk, +) +from yuxi.channel.extensions.nextcloud_talk.types import NextcloudTalkInboundMessage +from yuxi.channel.extensions.nextcloud_talk.webhook_server import ( + NextcloudTalkWebhookHandler, + create_nextcloud_talk_webhook_app, +) +from yuxi.channel.message.models import GroupContext, MessageType, PeerInfo, UnifiedMessage +from yuxi.channel.routing.models import PeerKind + +logger = logging.getLogger(__name__) + + +class NextcloudTalkGateway: + def __init__(self): + self._runner: web.AppRunner | None = None + self._site: web.TCPSite | None = None + self._account: object | None = None + self._cfg: NextcloudTalkConfig | None = None + self._ctx: ChannelContext | None = None + + async def start(self, ctx: ChannelContext) -> object: + self._ctx = ctx + channel_cfg = ctx.config.get("channels", {}).get("nextcloud-talk", {}) + self._cfg = NextcloudTalkConfig(**channel_cfg) + self._account = resolve_nextcloud_talk_account(self._cfg) + + if not self._account.configured: + raise RuntimeError("Nextcloud Talk gateway not configured") + + replay_guard = ReplayGuard() + + room_resolver = RoomInfoResolver( + base_url=self._account.base_url, + api_user=self._account.api_user, + api_password=self._account.api_password, + ) + + async def _send_callback(room_token: str, text: str) -> dict: + return await send_message_nextcloud_talk( + to=room_token, + text=text, + base_url=self._account.base_url, + secret=self._account.bot_secret, + dangerously_allow_private_network=self._cfg.dangerously_allow_private_network, + ) + + async def _inbound_handler(message: NextcloudTalkInboundMessage): + result = await handle_nextcloud_talk_inbound( + message=message, + cfg=self._cfg, + room_resolver=room_resolver, + ) + if result is None: + return None + + if result.get("action") == "pairing_challenge": + sender_id = result["sender_id"] + sender_name = result.get("sender_name", sender_id) + room_token = result["room_token"] + logger.info( + "Nextcloud Talk: pairing challenge for sender '%s'", + sender_id, + ) + try: + await _send_callback( + room_token, + f"Hi {sender_name}! Your ForcePilot access requires approval. " + f"Your Nextcloud user id: {sender_id}. " + f"Please ask an admin to add you to the allowlist.", + ) + except NextcloudTalkSendError as e: + logger.warning("Nextcloud Talk: failed to send pairing message: %s", e) + return None + + unified_msg = _to_unified_message(message, result, ctx) + if ctx.queue is not None: + await ctx.queue.put(unified_msg) + return result + + handler = NextcloudTalkWebhookHandler( + secret=self._account.bot_secret, + replay_guard=replay_guard, + base_url=self._account.base_url, + inbound_handler=_inbound_handler, + send_callback=_send_callback, + auto_join_message=self._cfg.auto_join_message, + ) + + app = create_nextcloud_talk_webhook_app(handler, webhook_path=self._cfg.webhook_path) + + self._runner = web.AppRunner(app) + await self._runner.setup() + + self._site = web.TCPSite( + self._runner, + host=self._cfg.webhook_host, + port=self._cfg.webhook_port, + ) + await self._site.start() + + public_url = ( + self._cfg.webhook_public_url + or f"http://{self._cfg.webhook_host}:{self._cfg.webhook_port}{self._cfg.webhook_path}" + ) + logger.info( + "Nextcloud Talk webhook listening on %s:%d (public: %s)", + self._cfg.webhook_host, + self._cfg.webhook_port, + public_url, + ) + + await ctx.cancel_event.wait() + + await self._cleanup() + + async def stop(self, ctx: ChannelContext) -> None: + ctx.cancel_event.set() + await self._cleanup() + + async def _cleanup(self) -> None: + if self._site: + await self._site.stop() + self._site = None + + if self._runner: + await self._runner.cleanup() + self._runner = None + + logger.info("Nextcloud Talk webhook stopped") + + async def check_health(self) -> bool: + return self._site is not None + + @property + def account(self): + return self._account + + @property + def cfg(self) -> NextcloudTalkConfig | None: + return self._cfg + + +def _to_unified_message( + message: NextcloudTalkInboundMessage, + inbound_result: dict, + ctx: ChannelContext, +) -> UnifiedMessage: + chat_type = inbound_result.get("ChatType", "group") + peer_kind = PeerKind.DIRECT if chat_type == "direct" else PeerKind.GROUP + + sender = PeerInfo( + kind=peer_kind, + id=message.sender_id, + display_name=message.sender_name, + ) + + group = None + if chat_type == "group": + group = GroupContext( + id=message.room_token, + name=message.room_name, + ) + + metadata = { + "room_token": message.room_token, + "room_name": message.room_name, + "sender_name": message.sender_name, + "media_type": message.media_type, + "content_format": "markdown" if message.media_type == "text/markdown" else "plain", + "was_mentioned": inbound_result.get("WasMentioned", False), + } + if message.reply_to_id: + metadata["reply_to_id"] = message.reply_to_id + if message.participant_type is not None: + metadata["participant_type"] = message.participant_type + + group_system_prompt = inbound_result.get("GroupSystemPrompt") + if group_system_prompt: + metadata["group_system_prompt"] = group_system_prompt + + return UnifiedMessage( + msg_id=message.message_id, + channel_type="nextcloud-talk", + account_id=ctx.account_id, + content=message.text, + sender=sender, + message_type=MessageType.TEXT, + group=group, + timestamp=datetime.now(UTC), + metadata=metadata, + conversation_label=inbound_result.get("ConversationLabel"), + ) diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/inbound.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/inbound.py new file mode 100644 index 00000000..9b6c62fb --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/inbound.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.nextcloud_talk.config_schema import ( + DmPolicy, + GroupPolicy, + NextcloudTalkConfig, + NextcloudTalkRoomConfig, +) +from yuxi.channel.extensions.nextcloud_talk.policy import ( + check_nextcloud_talk_group_sender_allowlist, + check_nextcloud_talk_sender_allowlist, + resolve_nextcloud_talk_room_match, +) +from yuxi.channel.extensions.nextcloud_talk.room_info import RoomInfoResolver +from yuxi.channel.extensions.nextcloud_talk.types import NextcloudTalkInboundMessage + +logger = logging.getLogger(__name__) + + +async def handle_nextcloud_talk_inbound( + message: NextcloudTalkInboundMessage, + cfg: NextcloudTalkConfig, + room_resolver: RoomInfoResolver, +) -> dict | None: + if not message.text.strip(): + logger.debug("Nextcloud Talk: empty message, discarding") + return None + + room_kind = await room_resolver.resolve(message.room_token) + if room_kind is None: + room_kind = "group" + + room_cfg = resolve_nextcloud_talk_room_match(cfg.rooms, message.room_token) + + if room_kind == "direct": + return await _handle_direct_message(message, cfg) + else: + return await _handle_group_message(message, cfg, room_cfg) + + +async def _handle_direct_message( + message: NextcloudTalkInboundMessage, + cfg: NextcloudTalkConfig, +) -> dict | None: + if cfg.dm_policy == DmPolicy.DISABLED: + logger.debug("Nextcloud Talk: DM disabled, discarding") + return None + + if cfg.dm_policy == DmPolicy.ALLOWLIST: + if not check_nextcloud_talk_sender_allowlist(message.sender_id, cfg): + logger.debug("Nextcloud Talk: sender not in DM allowlist") + return None + + if cfg.dm_policy == DmPolicy.PAIRING: + allowed = _is_sender_in_list(message.sender_id, cfg.allow_from) + if not allowed: + logger.info( + "Nextcloud Talk: sender '%s' not paired, initiating pairing challenge", + message.sender_id, + ) + return { + "action": "pairing_challenge", + "sender_id": message.sender_id, + "sender_name": message.sender_name, + "room_token": message.room_token, + } + + return _build_context_payload(message, "direct", was_mentioned=False) + + +async def _handle_group_message( + message: NextcloudTalkInboundMessage, + cfg: NextcloudTalkConfig, + room_cfg: NextcloudTalkRoomConfig | None, +) -> dict | None: + if cfg.group_policy == GroupPolicy.DISABLED: + logger.debug("Nextcloud Talk: group disabled, discarding") + return None + + if cfg.group_policy == GroupPolicy.ALLOWLIST: + if room_cfg is None: + logger.debug("Nextcloud Talk: room '%s' not in allowlist", message.room_token) + return None + if not room_cfg.enabled: + logger.debug("Nextcloud Talk: room '%s' disabled", message.room_token) + return None + if not check_nextcloud_talk_group_sender_allowlist(message.sender_id, cfg): + logger.debug("Nextcloud Talk: sender '%s' not in group allowlist", message.sender_id) + return None + + if room_cfg and not room_cfg.enabled: + logger.debug("Nextcloud Talk: room '%s' disabled", message.room_token) + return None + + require_mention = room_cfg.require_mention if room_cfg else True + was_mentioned = _check_mention(message.text) + + if require_mention and not was_mentioned: + logger.debug("Nextcloud Talk: require_mention set, but no @mention found") + return None + + return _build_context_payload(message, "group", was_mentioned=was_mentioned, room_cfg=room_cfg) + + +def _build_context_payload( + message: NextcloudTalkInboundMessage, + chat_type: str, + *, + was_mentioned: bool = False, + room_cfg: NextcloudTalkRoomConfig | None = None, +) -> dict: + payload = { + "From": f"nextcloud-talk:{message.sender_id}", + "To": f"nextcloud-talk:{message.room_token}", + "ChatType": chat_type, + "ConversationLabel": message.sender_name if chat_type == "direct" else f"room:{message.room_name}", + "SenderId": message.sender_id, + "SenderName": message.sender_name, + "Text": message.text, + "MessageId": message.message_id, + "Provider": "nextcloud-talk", + "Surface": "nextcloud-talk", + } + + if chat_type == "group": + payload["WasMentioned"] = was_mentioned + payload["GroupSubject"] = message.room_name + payload["From"] = f"nextcloud-talk:room:{message.room_token}" + if room_cfg and room_cfg.system_prompt: + payload["GroupSystemPrompt"] = room_cfg.system_prompt + + return payload + + +def _check_mention(text: str) -> bool: + import re + + return bool(re.search(r"@\S+", text)) + + +def _is_sender_in_list(sender_id: str, allow_list: list[str]) -> bool: + if not allow_list: + return False + normalized = sender_id.strip().lower() + for prefix in ("nextcloud-talk:", "nc-talk:", "nc:"): + if normalized.startswith(prefix): + normalized = normalized[len(prefix) :] + break + for entry in allow_list: + entry_normalized = entry.strip().lower() + for prefix in ("nextcloud-talk:", "nc-talk:", "nc:"): + if entry_normalized.startswith(prefix): + entry_normalized = entry_normalized[len(prefix) :] + break + if normalized == entry_normalized: + return True + return False diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/normalize.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/normalize.py new file mode 100644 index 00000000..8a375397 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/normalize.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +import re + +CHANNEL_PREFIXES = ("nextcloud-talk:", "nc-talk:", "nc:") + + +def strip_nextcloud_talk_target_prefix(target: str) -> str: + result = target.strip() + for prefix in CHANNEL_PREFIXES: + if result.lower().startswith(prefix.lower()): + result = result[len(prefix) :] + break + if result.startswith("room:"): + result = result[5:] + return result + + +def normalize_nextcloud_talk_messaging_target(target: str) -> str: + token = strip_nextcloud_talk_target_prefix(target) + return f"nextcloud-talk:{token}" + + +def looks_like_nextcloud_talk_target_id(target: str) -> bool: + trimmed = target.strip() + if re.match(r"^(nextcloud-talk|nc-talk|nc):", trimmed, re.IGNORECASE): + return True + return bool(re.match(r"^[a-z0-9]{8,}$", trimmed, re.IGNORECASE)) diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/plugin.json b/backend/package/yuxi/channel/extensions/nextcloud_talk/plugin.json new file mode 100644 index 00000000..afc12810 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/plugin.json @@ -0,0 +1,23 @@ +{ + "id": "nextcloud-talk", + "name": "Nextcloud Talk", + "version": "1.0.0", + "label": "Nextcloud Talk (Webhook Bot)", + "aliases": ["nc-talk", "nc"], + "description": "Self-hosted chat via Nextcloud Talk webhook bots.", + "author": "ForcePilot", + "order": 65, + "enabled": false, + "dependencies": ["httpx", "aiohttp"], + "capabilities": { + "chat_types": ["direct", "group"], + "message_types": ["text"], + "reactions": true, + "threads": false, + "media": true, + "polls": true, + "native_commands": false, + "block_streaming": true, + "streaming_mode": "block" + } +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/policy.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/policy.py new file mode 100644 index 00000000..9eb078d1 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/policy.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import logging + +from yuxi.channel.extensions.nextcloud_talk.config_schema import NextcloudTalkConfig, NextcloudTalkRoomConfig + +logger = logging.getLogger(__name__) + + +def resolve_nextcloud_talk_room_match( + rooms: dict[str, NextcloudTalkRoomConfig], + room_token: str, +) -> NextcloudTalkRoomConfig | None: + if room_token in rooms: + return rooms[room_token] + + lower_token = room_token.lower() + for key, config in rooms.items(): + if key.lower() == lower_token: + return config + + if "*" in rooms: + return rooms["*"] + + return None + + +def check_nextcloud_talk_sender_allowlist( + sender_id: str, + cfg: NextcloudTalkConfig, +) -> bool: + if not cfg.allow_from: + return True + normalized_sender = _normalize_sender_id(sender_id) + return normalized_sender in [_normalize_sender_id(s) for s in cfg.allow_from] + + +def check_nextcloud_talk_group_sender_allowlist( + sender_id: str, + cfg: NextcloudTalkConfig, +) -> bool: + if not cfg.group_allow_from: + return True + normalized_sender = _normalize_sender_id(sender_id) + return normalized_sender in [_normalize_sender_id(s) for s in cfg.group_allow_from] + + +def _normalize_sender_id(sender_id: str) -> str: + sid = sender_id.strip().lower() + for prefix in ("nextcloud-talk:", "nc-talk:", "nc:"): + if sid.startswith(prefix): + sid = sid[len(prefix) :] + break + return sid diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/poll_api.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/poll_api.py new file mode 100644 index 00000000..81c4b095 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/poll_api.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import base64 +import logging +from urllib.parse import urljoin + +import httpx + +logger = logging.getLogger(__name__) + +OCS_POLL_PATH = "/ocs/v2.php/apps/spreed/api/v1/poll/{token}" +OCS_POLL_ID_PATH = "/ocs/v2.php/apps/spreed/api/v1/poll/{token}/{poll_id}" + +DEFAULT_TIMEOUT = 30.0 + + +def _auth_headers(api_user: str, api_password: str) -> dict: + auth = base64.b64encode(f"{api_user}:{api_password}".encode()).decode() + return { + "Authorization": f"Basic {auth}", + "OCS-APIRequest": "true", + "Accept": "application/json", + } + + +async def create_poll( + room_token: str, + question: str, + options: list[str], + base_url: str, + api_user: str, + api_password: str, + *, + result_mode: int = 0, + max_votes: int = 1, + timeout: float = DEFAULT_TIMEOUT, +) -> dict | None: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_POLL_PATH.format(token=room_token), + ) + + body = { + "question": question, + "options": options, + "resultMode": result_mode, + "maxVotes": max_votes, + } + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + json=body, + headers=_auth_headers(api_user, api_password), + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk create poll failed: HTTP %s", resp.status_code) + return None + + data = resp.json() + return data.get("ocs", {}).get("data") + + +async def vote_poll( + room_token: str, + poll_id: str, + option_ids: list[int], + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> dict | None: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_POLL_ID_PATH.format(token=room_token, poll_id=poll_id), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + json={"options": option_ids}, + headers=_auth_headers(api_user, api_password), + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk vote poll failed: HTTP %s", resp.status_code) + return None + + data = resp.json() + return data.get("ocs", {}).get("data") + + +async def close_poll( + room_token: str, + poll_id: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> dict | None: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_POLL_ID_PATH.format(token=room_token, poll_id=poll_id), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.delete( + url, + headers=_auth_headers(api_user, api_password), + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk close poll failed: HTTP %s", resp.status_code) + return None + + data = resp.json() + return data.get("ocs", {}).get("data") diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/probe.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/probe.py new file mode 100644 index 00000000..e9d2dcd4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/probe.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +import base64 +import logging +import time +from dataclasses import dataclass, field +from urllib.parse import urljoin + +import httpx + +logger = logging.getLogger(__name__) + +OCS_CAPABILITIES_PATH = "/ocs/v2.php/cloud/capabilities" +PROBE_TIMEOUT = 10.0 + + +@dataclass +class NextcloudTalkProbeResult: + ok: bool = False + base_url: str = "" + reachable: bool = False + latency_ms: float | None = None + nextcloud_version: str | None = None + talk_app_installed: bool = False + bots_v1_available: bool = False + reactions_available: bool = False + api_credentials_valid: bool | None = None + # --feature response cannot be verified without a valid room token + feature_response_checkable: bool = False + warnings: list[str] = field(default_factory=list) + error: str | None = None + + +async def probe_nextcloud_talk_connectivity(base_url: str) -> dict: + if not base_url: + return {"reachable": False, "error": "base_url is not configured"} + + url = urljoin(base_url.rstrip("/") + "/", OCS_CAPABILITIES_PATH) + start = time.monotonic() + + try: + async with httpx.AsyncClient(timeout=PROBE_TIMEOUT) as client: + resp = await client.get( + url, + headers={ + "OCS-APIRequest": "true", + "Accept": "application/json", + }, + ) + latency_ms = (time.monotonic() - start) * 1000 + + if resp.status_code != 200: + return {"reachable": False, "latency_ms": latency_ms, "error": f"HTTP {resp.status_code}"} + + data = resp.json() + ocs_data = data.get("ocs", {}).get("data", {}) + version = ocs_data.get("version", {}).get("string") + capabilities = ocs_data.get("capabilities", {}) + talk_installed = "spreed" in capabilities + talk_capabilities = capabilities.get("spreed", {}).get("features", []) + + return { + "reachable": True, + "latency_ms": round(latency_ms, 1), + "nextcloud_version": version, + "talk_app_installed": talk_installed, + "bots_v1_available": "bots-v1" in talk_capabilities, + "reactions_available": "reactions" in talk_capabilities, + } + except httpx.TimeoutException: + return {"reachable": False, "error": "connection timeout"} + except httpx.ConnectError as e: + return {"reachable": False, "error": f"connection failed: {e}"} + except Exception as e: + return {"reachable": False, "error": str(e)} + + +async def probe_nextcloud_talk_api_credentials(base_url: str, api_user: str, api_password: str) -> dict: + if not api_user or not api_password: + return {"valid": None, "error": "API credentials not configured"} + + url = urljoin(base_url.rstrip("/") + "/", OCS_CAPABILITIES_PATH) + auth = base64.b64encode(f"{api_user}:{api_password}".encode()).decode() + + try: + async with httpx.AsyncClient(timeout=PROBE_TIMEOUT) as client: + resp = await client.get( + url, + headers={ + "Authorization": f"Basic {auth}", + "OCS-APIRequest": "true", + "Accept": "application/json", + }, + ) + if resp.status_code == 200: + return {"valid": True} + if resp.status_code == 401: + return {"valid": False, "error": "API credentials invalid (401)"} + return {"valid": False, "error": f"HTTP {resp.status_code}"} + except Exception as e: + return {"valid": False, "error": str(e)} + + +async def probe_nextcloud_talk(account: dict) -> NextcloudTalkProbeResult: + base_url = account.get("base_url", "") + api_user = account.get("api_user") + api_password = account.get("api_password", "") + bot_secret = account.get("bot_secret", "") + + result = NextcloudTalkProbeResult(base_url=base_url) + + if not base_url: + result.error = "base_url not configured" + result.warnings.append("base_url is required") + return result + + conn_result = await probe_nextcloud_talk_connectivity(base_url) + result.reachable = conn_result["reachable"] + result.latency_ms = conn_result.get("latency_ms") + result.nextcloud_version = conn_result.get("nextcloud_version") + result.talk_app_installed = conn_result.get("talk_app_installed", False) + result.bots_v1_available = conn_result.get("bots_v1_available", False) + result.reactions_available = conn_result.get("reactions_available", False) + + if not result.reachable: + result.error = conn_result.get("error", "unreachable") + result.warnings.append(f"Nextcloud instance unreachable: {result.error}") + return result + + if not result.talk_app_installed: + result.warnings.append("Nextcloud Talk app not detected on this instance") + + if not result.bots_v1_available: + result.warnings.append("bots-v1 capability not available - Nextcloud Talk may be too old for Bot API") + + if api_user and api_password: + cred_result = await probe_nextcloud_talk_api_credentials(base_url, api_user, api_password) + result.api_credentials_valid = cred_result["valid"] + if not cred_result["valid"]: + result.warnings.append(f"API credentials: {cred_result.get('error', 'unknown error')}") + + if not bot_secret: + result.warnings.append("bot_secret not configured - webhook signature verification will fail") + + result.feature_response_checkable = bool(bot_secret) + if not result.feature_response_checkable: + result.warnings.append( + "Cannot verify --feature response: bot_secret not configured. " + "Ensure 'occ talk:bot:install --feature response' was used." + ) + + result.ok = result.reachable and not any(w for w in result.warnings if "unreachable" in w.lower()) + + return result diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/reaction_api.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/reaction_api.py new file mode 100644 index 00000000..b05ce79a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/reaction_api.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import base64 +import logging +from urllib.parse import urljoin + +import httpx + +logger = logging.getLogger(__name__) + +OCS_REACTION_PATH = "/ocs/v2.php/apps/spreed/api/v1/reaction/{token}/{message_id}" + +DEFAULT_TIMEOUT = 30.0 + + +def _auth_headers(api_user: str, api_password: str) -> dict: + auth = base64.b64encode(f"{api_user}:{api_password}".encode()).decode() + return { + "Authorization": f"Basic {auth}", + "OCS-APIRequest": "true", + "Accept": "application/json", + } + + +async def list_reactions( + room_token: str, + message_id: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> dict: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_REACTION_PATH.format(token=room_token, message_id=message_id), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.get( + url, + headers=_auth_headers(api_user, api_password), + ) + + if resp.status_code != 200: + logger.warning("Nextcloud Talk list reactions failed: HTTP %s", resp.status_code) + return {} + + data = resp.json() + return data.get("ocs", {}).get("data", {}) + + +async def add_reaction_user( + room_token: str, + message_id: str, + reaction: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_REACTION_PATH.format(token=room_token, message_id=message_id), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + json={"reaction": reaction}, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 + + +async def remove_reaction_user( + room_token: str, + message_id: str, + reaction: str, + base_url: str, + api_user: str, + api_password: str, + *, + timeout: float = DEFAULT_TIMEOUT, +) -> bool: + url = urljoin( + base_url.rstrip("/") + "/", + OCS_REACTION_PATH.format(token=room_token, message_id=message_id), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.delete( + url, + json={"reaction": reaction}, + headers=_auth_headers(api_user, api_password), + ) + + return resp.status_code == 200 diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/replay_guard.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/replay_guard.py new file mode 100644 index 00000000..2565a180 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/replay_guard.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import time +from collections import OrderedDict +from pathlib import Path + +logger = logging.getLogger(__name__) + +DEFAULT_REPLAY_TTL_S = 24 * 60 * 60 +DEFAULT_MEMORY_MAX_SIZE = 1000 + + +class ReplayGuard: + def __init__( + self, + account_id: str = "default", + state_dir: Path | None = None, + ttl_s: int = DEFAULT_REPLAY_TTL_S, + max_memory: int = DEFAULT_MEMORY_MAX_SIZE, + ): + self.account_id = account_id + self.ttl_s = ttl_s + self._claimed: dict[str, float] = {} + self._committed: OrderedDict[str, float] = OrderedDict() + self._max_memory = max_memory + self._state_file = state_dir / "nextcloud-talk" / f"replay-{account_id}.json" if state_dir else None + self._lock = asyncio.Lock() + + def _build_key(self, room_token: str, message_id: str) -> str | None: + room_token = room_token.strip() + message_id = message_id.strip() + if not room_token or not message_id: + return None + return f"{room_token}:{message_id}" + + async def claim_message(self, room_token: str, message_id: str) -> str: + key = self._build_key(room_token, message_id) + if key is None: + return "invalid" + + now = time.monotonic() + async with self._lock: + self._gc_locked(now) + + if key in self._committed: + return "duplicate" + + if key in self._claimed: + return "inflight" + + self._claimed[key] = now + return "claimed" + + async def commit_message(self, room_token: str, message_id: str) -> None: + key = self._build_key(room_token, message_id) + if key is None: + return + + async with self._lock: + self._claimed.pop(key, None) + self._committed[key] = time.monotonic() + + while len(self._committed) > self._max_memory: + self._committed.popitem(last=False) + + await self._maybe_persist() + + async def release_message(self, room_token: str, message_id: str) -> None: + key = self._build_key(room_token, message_id) + if key is None: + return + + async with self._lock: + self._claimed.pop(key, None) + + def _gc_locked(self, now: float) -> None: + threshold = now - self.ttl_s + + expired_claimed = [k for k, t in self._claimed.items() if t < threshold] + for k in expired_claimed: + del self._claimed[k] + + while self._committed: + _, t = next(iter(self._committed.items())) + if t >= threshold: + break + self._committed.popitem(last=False) + + async def _maybe_persist(self) -> None: + if self._state_file is None: + return + try: + self._state_file.parent.mkdir(parents=True, exist_ok=True) + data = { + "committed": dict(self._committed), + "saved_at": time.time(), + } + content = json.dumps(data) + await asyncio.to_thread(self._state_file.write_text, content, encoding="utf-8") + except Exception as e: + logger.warning("ReplayGuard persist failed: %s", e) diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/room_info.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/room_info.py new file mode 100644 index 00000000..2b9a11f5 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/room_info.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import base64 +import logging +import time +from urllib.parse import urljoin + +import httpx + +logger = logging.getLogger(__name__) + +ROOM_CACHE_TTL_S = 5 * 60 +ROOM_CACHE_ERROR_TTL_S = 30 + +OCS_ROOM_PATH = "/ocs/v2.php/apps/spreed/api/v4/room/{token}" + +ROOM_TYPE_DIRECT = {1, 5, 6} + + +class RoomCacheEntry: + __slots__ = ("kind", "fetched_at", "error") + + def __init__(self, kind: str | None = None, error: str | None = None): + self.kind = kind + self.fetched_at = time.monotonic() + self.error = error + + +class RoomInfoResolver: + def __init__(self, base_url: str, api_user: str | None, api_password: str | None): + self.base_url = base_url + self.api_user = api_user + self.api_password = api_password + self._cache: dict[str, RoomCacheEntry] = {} + + @property + def available(self) -> bool: + return bool(self.api_user and self.api_password) + + def _cache_key(self, room_token: str) -> str: + return room_token + + def _get_cached(self, room_token: str) -> str | None: + entry = self._cache.get(self._cache_key(room_token)) + if entry is None: + return None + ttl = ROOM_CACHE_ERROR_TTL_S if entry.error else ROOM_CACHE_TTL_S + if time.monotonic() - entry.fetched_at > ttl: + del self._cache[self._cache_key(room_token)] + return None + return entry.kind + + async def resolve(self, room_token: str) -> str | None: + cached = self._get_cached(room_token) + if cached is not None: + return cached + + if not self.available: + return None + + url = urljoin( + self.base_url.rstrip("/") + "/", + OCS_ROOM_PATH.format(token=room_token), + ) + + try: + auth = base64.b64encode(f"{self.api_user}:{self.api_password}".encode()).decode() + + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get( + url, + headers={ + "Authorization": f"Basic {auth}", + "OCS-APIRequest": "true", + "Accept": "application/json", + }, + ) + + if resp.status_code != 200: + self._cache[self._cache_key(room_token)] = RoomCacheEntry(error=f"HTTP {resp.status_code}") + return None + + data = resp.json() + room_type = data.get("ocs", {}).get("data", {}).get("type") + + if room_type is None: + self._cache[self._cache_key(room_token)] = RoomCacheEntry(error="no type field") + return None + + kind = "direct" if room_type in ROOM_TYPE_DIRECT else "group" + self._cache[self._cache_key(room_token)] = RoomCacheEntry(kind=kind) + return kind + + except Exception as e: + logger.warning("Nextcloud Talk room-info query failed: %s", e) + self._cache[self._cache_key(room_token)] = RoomCacheEntry(error=str(e)) + return None diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/send.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/send.py new file mode 100644 index 00000000..ff94112a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/send.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +import ipaddress +import logging +import secrets +from urllib.parse import urljoin, urlparse + +import httpx + +from yuxi.channel.extensions.nextcloud_talk.format import convert_markdown_tables_to_ascii +from yuxi.channel.extensions.nextcloud_talk.normalize import strip_nextcloud_talk_target_prefix +from yuxi.channel.extensions.nextcloud_talk.signature import generate_nextcloud_talk_signature + +logger = logging.getLogger(__name__) + +OCS_MESSAGE_PATH = "/ocs/v2.php/apps/spreed/api/v1/bot/{token}/message" +OCS_REACTION_PATH = "/ocs/v2.php/apps/spreed/api/v1/bot/{token}/reaction/{message_id}" + + +class NextcloudTalkSendError(Exception): + def __init__(self, message: str, status_code: int | None = None): + super().__init__(message) + self.status_code = status_code + + +def _check_ssrf(base_url: str, dangerously_allow_private_network: bool = False) -> None: + if dangerously_allow_private_network: + return + + try: + parsed = urlparse(base_url) + except ValueError: + return + + hostname = parsed.hostname + if not hostname: + return + + from yuxi.channel.security.ssrf_guard import SsrfGuard + + guard = SsrfGuard(default_deny=False) + + try: + ip_str = str(ipaddress.ip_address(hostname)) + except ValueError: + return + + ip_result = guard.check_ip(ip_str) + if not ip_result.safe: + raise NextcloudTalkSendError( + f"SSRF blocked: {hostname} — {ip_result.reason}. Set dangerously_allow_private_network=true to allow." + ) + + +async def send_message_nextcloud_talk( + to: str, + text: str, + base_url: str, + secret: str, + *, + reply_to: str | None = None, + silent: bool = False, + reference_id: str | None = None, + timeout: float = 30.0, + dangerously_allow_private_network: bool = False, +) -> dict: + _check_ssrf(base_url, dangerously_allow_private_network) + + room_token = strip_nextcloud_talk_target_prefix(to) + + if not text.strip(): + raise NextcloudTalkSendError("Message must be non-empty") + + text = convert_markdown_tables_to_ascii(text) + + body = {"message": text} + if reply_to: + body["replyTo"] = reply_to + if silent: + body["silent"] = True + if reference_id is None: + reference_id = secrets.token_hex(16) + body["referenceId"] = reference_id + + random, signature = generate_nextcloud_talk_signature(text, secret) + + url = urljoin(base_url.rstrip("/") + "/", OCS_MESSAGE_PATH.format(token=room_token)) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + json=body, + headers={ + "Content-Type": "application/json", + "OCS-APIRequest": "true", + "X-Nextcloud-Talk-Bot-Random": random, + "X-Nextcloud-Talk-Bot-Signature": signature, + "Accept": "application/json", + }, + ) + + if resp.status_code == 200: + data = resp.json() + ocs_data = data.get("ocs", {}).get("data", {}) + return { + "message_id": str(ocs_data.get("id", "unknown")), + "room_token": room_token, + "timestamp": ocs_data.get("timestamp"), + "reference_id": reference_id, + } + + error_body = resp.text[:500] + status_map = { + 400: f"Nextcloud Talk: bad request - {error_body}", + 401: "Nextcloud Talk: authentication failed - check bot secret", + 403: "Nextcloud Talk: forbidden - bot may not have permission in this room", + 404: f"Nextcloud Talk: room not found (token={room_token})", + } + msg = status_map.get(resp.status_code, f"Nextcloud Talk send failed: {error_body}") + raise NextcloudTalkSendError(msg, resp.status_code) + + +async def send_reaction_nextcloud_talk( + room_token: str, + message_id: str, + reaction: str, + base_url: str, + secret: str, + *, + timeout: float = 30.0, + dangerously_allow_private_network: bool = False, +) -> bool: + _check_ssrf(base_url, dangerously_allow_private_network) + + random, signature = generate_nextcloud_talk_signature(reaction, secret) + + url = urljoin( + base_url.rstrip("/") + "/", + OCS_REACTION_PATH.format(token=room_token, message_id=message_id), + ) + + async with httpx.AsyncClient(timeout=timeout) as client: + resp = await client.post( + url, + json={"reaction": reaction}, + headers={ + "Content-Type": "application/json", + "OCS-APIRequest": "true", + "X-Nextcloud-Talk-Bot-Random": random, + "X-Nextcloud-Talk-Bot-Signature": signature, + "Accept": "application/json", + }, + ) + + return resp.status_code == 200 diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/signature.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/signature.py new file mode 100644 index 00000000..6853341f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/signature.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import hashlib +import hmac +import secrets + + +def verify_nextcloud_talk_signature( + signature: str, + random: str, + body: str, + secret: str, +) -> bool: + expected = hmac.new( + secret.encode("utf-8"), + (random + body).encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return hmac.compare_digest(expected, signature) + + +def generate_nextcloud_talk_signature( + body: str, + secret: str, +) -> tuple[str, str]: + random = secrets.token_hex(32) + signature = hmac.new( + secret.encode("utf-8"), + (random + body).encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return random, signature + + +def extract_nextcloud_talk_headers( + headers: dict, +) -> dict | None: + signature = headers.get("X-Nextcloud-Talk-Signature") + random = headers.get("X-Nextcloud-Talk-Random") + backend = headers.get("X-Nextcloud-Talk-Backend") + + if not all([signature, random, backend]): + return None + + return { + "signature": signature, + "random": random, + "backend": backend, + } diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/types.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/types.py new file mode 100644 index 00000000..f2b18dff --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/types.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class NextcloudTalkWebhookPayload: + type: str + actor: dict + object: dict + target: dict + + +@dataclass +class NextcloudTalkInboundMessage: + message_id: str + room_token: str + room_name: str + sender_id: str + sender_name: str + text: str + media_type: str + timestamp: int + reply_to_id: str | None = None + participant_type: int | None = None + + +@dataclass +class NextcloudTalkSendResult: + message_id: str + room_token: str + timestamp: int | None = None + + +@dataclass +class NextcloudTalkHeaders: + signature: str + random: str + backend: str + + +@dataclass +class ResolvedNextcloudTalkAccount: + account_id: str = "default" + enabled: bool = True + name: str = "" + configured: bool = False + base_url: str = "" + bot_secret: str = "" + secret_source: str = "none" + api_user: str | None = None + api_password: str = "" + api_password_source: str = "none" + + +@dataclass +class RoomCacheEntry: + kind: str | None = None + fetched_at: float = 0.0 + error: str | None = None diff --git a/backend/package/yuxi/channel/extensions/nextcloud_talk/webhook_server.py b/backend/package/yuxi/channel/extensions/nextcloud_talk/webhook_server.py new file mode 100644 index 00000000..8728ef42 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/nextcloud_talk/webhook_server.py @@ -0,0 +1,251 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import time +from urllib.parse import urlparse + +from aiohttp import web + +from yuxi.channel.extensions.nextcloud_talk.replay_guard import ReplayGuard +from yuxi.channel.extensions.nextcloud_talk.signature import ( + extract_nextcloud_talk_headers, + verify_nextcloud_talk_signature, +) +from yuxi.channel.extensions.nextcloud_talk.types import NextcloudTalkInboundMessage + +logger = logging.getLogger(__name__) + + +def _resolve_rich_object_content(raw: str) -> str: + try: + parsed = json.loads(raw) + except (json.JSONDecodeError, TypeError): + return raw + + if not isinstance(parsed, dict): + return raw + + message = parsed.get("message") + if not message: + return raw + + parameters = parsed.get("parameters", {}) + if not parameters: + return message + + for key, value in parameters.items(): + placeholder = "{" + key + "}" + if isinstance(value, dict): + name = value.get("name") or value.get("id") or key + else: + name = str(value) + message = message.replace(placeholder, name) + + return message + + +DEFAULT_WEBHOOK_PORT = 8788 +DEFAULT_WEBHOOK_HOST = "0.0.0.0" +DEFAULT_WEBHOOK_PATH = "/nextcloud-talk-webhook" +PREAUTH_MAX_BODY_BYTES = 64 * 1024 +PREAUTH_BODY_TIMEOUT_S = 5.0 + + +class RetryableWebhookError(Exception): + pass + + +class NextcloudTalkWebhookHandler: + def __init__( + self, + secret: str, + replay_guard: ReplayGuard, + base_url: str, + inbound_handler, + *, + send_callback=None, + auto_join_message: str | None = None, + ): + self.secret = secret + self.replay_guard = replay_guard + self.expected_backend_origin = self._normalize_origin(base_url) + self.inbound_handler = inbound_handler + self._send_callback = send_callback + self._auto_join_message = auto_join_message + + @staticmethod + def _normalize_origin(url: str) -> str | None: + if not url: + return None + try: + parsed = urlparse(url) + return f"{parsed.scheme}://{parsed.hostname}".lower() + except Exception: + return None + + def _is_backend_allowed(self, backend_url: str) -> bool: + if not self.expected_backend_origin: + return True + backend_origin = self._normalize_origin(backend_url) + return backend_origin == self.expected_backend_origin + + async def handle_health(self, request: web.Request) -> web.Response: + return web.Response(text="ok", status=200) + + async def handle_webhook(self, request: web.Request) -> web.Response: + headers = extract_nextcloud_talk_headers(request.headers) + if not headers: + return web.json_response({"error": "Missing signature headers"}, status=400) + + if not self._is_backend_allowed(headers["backend"]): + logger.warning("Nextcloud Talk webhook: backend origin mismatch") + return web.json_response({"error": "Invalid backend"}, status=401) + + try: + body = await asyncio.wait_for(request.text(), timeout=PREAUTH_BODY_TIMEOUT_S) + except TimeoutError: + return web.json_response({"error": "Body read timeout"}, status=408) + + if len(body) > PREAUTH_MAX_BODY_BYTES: + return web.json_response({"error": "Body too large"}, status=413) + + if not verify_nextcloud_talk_signature( + signature=headers["signature"], + random=headers["random"], + body=body, + secret=self.secret, + ): + logger.warning("Nextcloud Talk webhook: invalid signature") + return web.json_response({"error": "Invalid signature"}, status=401) + + try: + payload = json.loads(body) + except json.JSONDecodeError: + return web.json_response({"error": "Invalid JSON"}, status=400) + + event_type = payload.get("type") + + if event_type == "Create": + actor = payload.get("actor", {}) + if actor.get("type") == "Application": + return web.json_response({}, status=200) + actor_id = str(actor.get("id", "")) + if actor_id.startswith("bots/"): + return web.json_response({}, status=200) + + try: + message = self._parse_inbound_message(payload) + except (KeyError, ValueError) as e: + return web.json_response({"error": f"Invalid payload: {e}"}, status=400) + + claim = await self.replay_guard.claim_message( + room_token=message.room_token, + message_id=message.message_id, + ) + if claim != "claimed": + return web.json_response({}, status=200) + + asyncio.create_task(self._process_message(message)) + return web.json_response({}, status=200) + + if event_type in ("Like", "Undo"): + actor = payload.get("actor", {}) + obj = payload.get("object", {}) + target = payload.get("target", {}) + reaction = obj.get("content") or obj.get("name") or "" + message_id = str(obj.get("inReplyTo", obj.get("id", ""))) + room_token = str(target.get("id", "")) + actor_name = actor.get("name", "") + logger.info( + "Nextcloud Talk reaction %s: '%s' on message %s by %s in room %s", + event_type.lower(), + reaction, + message_id, + actor_name, + room_token, + ) + return web.json_response({}, status=200) + + if event_type == "Join": + target = payload.get("target", {}) + room_token = str(target.get("id", "")) + logger.info( + "Nextcloud Talk: bot joined room %s (room_token=%s)", + target.get("name", room_token), + room_token, + ) + if self._auto_join_message and self._send_callback: + asyncio.create_task(self._send_callback(room_token, self._auto_join_message)) + return web.json_response({}, status=200) + + if event_type == "Leave": + target = payload.get("target", {}) + logger.info( + "Nextcloud Talk: bot left room %s (room_token=%s)", + target.get("name", target.get("id", "")), + target.get("id", ""), + ) + return web.json_response({}, status=200) + + return web.json_response({}, status=200) + + def _parse_inbound_message(self, payload: dict) -> NextcloudTalkInboundMessage: + actor = payload["actor"] + obj = payload["object"] + target = payload["target"] + + raw_content = obj.get("content") or obj.get("name") or "" + text = _resolve_rich_object_content(raw_content) + media_type = obj.get("mediaType", "text/plain") + + reply_to_id = str(obj.get("inReplyTo", "")) or None + participant_type = actor.get("talkParticipantType") + + return NextcloudTalkInboundMessage( + message_id=str(obj["id"]), + room_token=str(target["id"]), + room_name=target.get("name", ""), + sender_id=str(actor["id"]), + sender_name=actor.get("name", ""), + text=text, + media_type=media_type, + timestamp=int(time.time() * 1000), + reply_to_id=reply_to_id, + participant_type=participant_type, + ) + + async def _process_message(self, message: NextcloudTalkInboundMessage) -> None: + try: + await self.inbound_handler(message) + await self.replay_guard.commit_message( + room_token=message.room_token, + message_id=message.message_id, + ) + except RetryableWebhookError: + await self.replay_guard.release_message( + room_token=message.room_token, + message_id=message.message_id, + ) + except Exception: + await self.replay_guard.commit_message( + room_token=message.room_token, + message_id=message.message_id, + ) + raise + + +def create_nextcloud_talk_webhook_app( + handler: NextcloudTalkWebhookHandler, + webhook_path: str = DEFAULT_WEBHOOK_PATH, +) -> web.Application: + app = web.Application() + app.router.add_get("/healthz", handler.handle_health) + app.router.add_post(webhook_path, handler.handle_webhook) + + async def _not_found(_request): + return web.json_response({"error": "Not found"}, status=404) + + app.router.add_route("*", "/{tail:.*}", _not_found) + return app