From b018ad25da56620714871d9e8b033011c8d27d9d Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Tue, 12 May 2026 00:47:06 +0800 Subject: [PATCH] feat(backend): add Nextcloud Talk channel adapter implementation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现了完整的Nextcloud Talk聊天适配器，包含以下核心功能： 1. 基础客户端通信与认证 2. 会话路由与聊天类型解析 3. 消息分块与格式转换 4. 用户配对与权限校验 5. 轮询式消息监听 6. 配置验证与诊断工具 7. 安全策略与去重缓存 8. 指标统计与状态快照 新增配套的配对用户缓存文件与模块导出入口。 --- .../adapters/nextcloudtalk/.paired_users.json | 1 + .../adapters/nextcloudtalk/__init__.py | 3 + .../adapters/nextcloudtalk/accounts.py | 82 ++ .../adapters/nextcloudtalk/adapter.py | 1018 +++++++++++++++++ .../adapters/nextcloudtalk/approval_auth.py | 54 + .../adapters/nextcloudtalk/chunking.py | 60 + .../channels/adapters/nextcloudtalk/client.py | 121 ++ .../channels/adapters/nextcloudtalk/config.py | 91 ++ .../adapters/nextcloudtalk/config_schema.py | 188 +++ .../adapters/nextcloudtalk/credentials.py | 57 + .../channels/adapters/nextcloudtalk/dedup.py | 127 ++ .../adapters/nextcloudtalk/directory.py | 103 ++ .../channels/adapters/nextcloudtalk/doctor.py | 187 +++ .../adapters/nextcloudtalk/formatter.py | 103 ++ .../adapters/nextcloudtalk/metrics.py | 79 ++ .../adapters/nextcloudtalk/monitor_polling.py | 163 +++ .../adapters/nextcloudtalk/monitor_ws.py | 210 ++++ .../adapters/nextcloudtalk/normalize.py | 59 + .../adapters/nextcloudtalk/normalizer.py | 221 ++++ .../adapters/nextcloudtalk/pairing.py | 75 ++ .../channels/adapters/nextcloudtalk/probe.py | 66 ++ .../adapters/nextcloudtalk/room_info.py | 70 ++ .../adapters/nextcloudtalk/secret_contract.py | 81 ++ .../adapters/nextcloudtalk/security.py | 209 ++++ .../channels/adapters/nextcloudtalk/send.py | 295 +++++ .../adapters/nextcloudtalk/session.py | 48 + .../adapters/nextcloudtalk/setup_wizard.py | 146 +++ .../channels/adapters/nextcloudtalk/status.py | 66 ++ .../adapters/nextcloudtalk/webhook_server.py | 229 ++++ 29 files changed, 4212 insertions(+) create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/.paired_users.json create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/accounts.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/adapter.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/approval_auth.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/chunking.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/client.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/config.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/config_schema.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/credentials.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/dedup.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/directory.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/doctor.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/formatter.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/metrics.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/monitor_polling.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/monitor_ws.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/normalize.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/normalizer.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/pairing.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/probe.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/room_info.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/secret_contract.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/security.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/send.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/session.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/setup_wizard.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/status.py create mode 100644 backend/package/yuxi/channels/adapters/nextcloudtalk/webhook_server.py diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/.paired_users.json b/backend/package/yuxi/channels/adapters/nextcloudtalk/.paired_users.json new file mode 100644 index 00000000..0637a088 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/.paired_users.json @@ -0,0 +1 @@ +[] \ No newline at end of file diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/__init__.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/__init__.py new file mode 100644 index 00000000..63812bcd --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/__init__.py @@ -0,0 +1,3 @@ +from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter + +__all__ = ["NextcloudTalkAdapter"] diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/accounts.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/accounts.py new file mode 100644 index 00000000..33467ff4 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/accounts.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import copy +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +DEFAULT_ACCOUNT_ID = "default" + + +class NcAccountConfig: + def __init__(self, account_id: str, config: dict[str, Any] | None = None): + self.account_id = account_id + self.config = config or {} + + @property + def server_url(self) -> str: + return self.config.get("server_url", self.config.get("baseUrl", "")) + + @property + def bot_user(self) -> str: + return self.config.get("bot_user", self.config.get("botUser", "")) + + @property + def app_password(self) -> str: + return self.config.get("app_password", self.config.get("appPassword", "")) + + @property + def display_name(self) -> str: + return self.config.get("display_name", self.config.get("displayName", "ForcePilot Bot")) + + @property + def dm_policy(self) -> str: + return self.config.get("dm_policy", self.config.get("dmPolicy", "pairing")) + + @property + def group_policy(self) -> str: + return self.config.get("group_policy", self.config.get("groupPolicy", "allowlist")) + + +def resolve_nc_account(account_id: str, channel_config: dict[str, Any]) -> NcAccountConfig | None: + accounts = channel_config.get("accounts", {}) + + if account_id == DEFAULT_ACCOUNT_ID and not accounts: + return NcAccountConfig(DEFAULT_ACCOUNT_ID, channel_config) + + account_cfg = accounts.get(account_id) + if not account_cfg: + return None + + merged = resolve_merged_config(account_id, channel_config) + return NcAccountConfig(account_id, merged) + + +def resolve_merged_config(account_id: str, channel_config: dict[str, Any]) -> dict[str, Any]: + base = {k: v for k, v in channel_config.items() if k not in ("accounts",)} + accounts = channel_config.get("accounts", {}) + account_cfg = accounts.get(account_id, {}) + + merged = copy.deepcopy(base) + _deep_merge(merged, account_cfg) + return merged + + +def _deep_merge(base: dict, override: dict) -> None: + for key, value in override.items(): + if key in base and isinstance(base[key], dict) and isinstance(value, dict): + _deep_merge(base[key], value) + else: + base[key] = value + + +def list_nc_accounts(channel_config: dict[str, Any]) -> list[str]: + accounts = channel_config.get("accounts", {}) + if not accounts: + return [DEFAULT_ACCOUNT_ID] + return [DEFAULT_ACCOUNT_ID] + [k for k in accounts if k != DEFAULT_ACCOUNT_ID] + + +def resolve_default_account_id(channel_config: dict[str, Any]) -> str: + return DEFAULT_ACCOUNT_ID diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/adapter.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/adapter.py new file mode 100644 index 00000000..fcb4a063 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/adapter.py @@ -0,0 +1,1018 @@ +from __future__ import annotations + +import asyncio +import logging +import os +import re +from collections.abc import AsyncIterator, Callable +from dataclasses import dataclass, field +from typing import Any, Literal +from urllib.parse import urlparse + +from yuxi.channels.base import BaseChannelAdapter +from yuxi.channels.capabilities import ChannelCapabilities +from yuxi.channels.meta import ChannelMeta +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.exceptions import ( + ChannelAuthenticationError, +) +from yuxi.channels.models import ( + ChannelMessage, + ChannelResponse, + ChannelStatus, + ChannelType, + ChatType, + DeliveryResult, + HealthStatus, +) +from yuxi.channels.registry import register_builtin_adapter +from yuxi.utils.datetime_utils import utc_now_naive + +from .client import NextcloudTalkClient, verify_hmac_signature +from .config import ( + resolve_room_enabled, + resolve_room_system_prompt, + resolve_room_skills, + resolve_dm_system_prompt, + resolve_dm_enabled, + resolve_dm_skills, +) +from .dedup import NextcloudTalkDedupGuard +from .formatter import format_outbound +from .metrics import NextcloudTalkMetrics +from .monitor_polling import PollingMonitor +from .monitor_ws import WebSocketMonitor, try_connect_websocket +from .normalizer import normalize_inbound +from .pairing import send_pairing_challenge +from .probe import probe_capabilities, resolve_api_base, resolve_api_version +from .security import check_dm_policy, check_group_policy, check_mention_gate, collect_security_warnings +from .send import ( + send_media, + send_reaction, + send_text, + send_reaction_delete, + get_reactions as _get_reactions, + edit_message as _send_edit, + delete_message as _send_delete, +) +from .session import resolve_session_route, make_thread_key + +logger = logging.getLogger(__name__) + +_DEBUG_ACCOUNTS = os.getenv("OPENCLAW_DEBUG_NEXTCLOUD_TALK_ACCOUNTS", "").strip().lower() in ( + "1", + "true", + "yes", + "on", +) + + +@dataclass +class _AccountSession: + account_id: str + label: str = "" + server_url: str = "" + bot_user: str = "" + app_password: str = "" + client: NextcloudTalkClient | None = None + api_base: str = "" + monitor_mode: Literal["websocket", "polling"] = "polling" + polling_monitor: PollingMonitor | None = None + ws_monitor: WebSocketMonitor | None = None + polling_task: asyncio.Task | None = None + ws_task: asyncio.Task | None = None + config: dict[str, Any] = field(default_factory=dict) + + +@register_builtin_adapter(aliases=["nc-talk", "nc", "nextcloud-talk"]) +class NextcloudTalkAdapter(BaseChannelAdapter): + channel_id = "nextcloud-talk" + channel_type = ChannelType.NEXTCLOUDTALK + + text_chunk_limit = 4000 + supports_markdown = True + supports_streaming = True + streaming_modes = ["off", "block"] + max_media_size_mb = 100 + + capabilities = ChannelCapabilities( + chat_types=["direct", "group"], + supports_markdown=True, + supports_streaming=True, + streaming_modes=["off", "block"], + text_chunk_limit=4000, + max_media_size_mb=100, + reactions=True, + edit=True, + unsend=True, + reply=True, + media=True, + block_streaming=True, + polls=True, + ) + meta = ChannelMeta( + id="nextcloud-talk", + label="Nextcloud Talk", + aliases=["nextcloudtalk", "nc-talk", "nc", "nextcloud-talk"], + ) + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self._status = ChannelStatus.DISCONNECTED + self._client: NextcloudTalkClient | None = None + self._monitor_mode: Literal["websocket", "polling"] = "polling" + self._api_base: str = "" + self._bot_display_name: str = "ForcePilot Bot" + self._conversations: dict[str, dict[str, Any]] = {} + self._circuit_breaker = CircuitBreaker(failure_threshold=5) + self._polling_monitor: PollingMonitor | None = None + self._ws_monitor: WebSocketMonitor | None = None + self._polling_task: asyncio.Task | None = None + self._ws_task: asyncio.Task | None = None + self._stream_msg_ids: dict[str, str] = {} + self._dedup_guard = NextcloudTalkDedupGuard() + self._sent_cache: dict[str, str] = {} + self._on_send_callback: Callable[[DeliveryResult], None] | None = None + self._stream_edit_times: dict[str, float] = {} + self._stream_msg_ttls: dict[str, float] = {} + self._stream_max_edits_per_sec = 3 + self._reaction_level: str = "minimal" + self._stream_coalesce: dict[str, Any] = {} + self._sessions: dict[str, _AccountSession] = {} + self._primary_account_id: str | None = None + self._is_multi_account: bool = False + + self._inbound_count: int = 0 + self._outbound_count: int = 0 + self._last_inbound_at: float | None = None + self._last_outbound_at: float | None = None + + self._metrics = NextcloudTalkMetrics() + + self._resolve_accounts() + + @property + def status(self) -> str: + return self._status.value + + async def connect(self) -> None: + if self._status == ChannelStatus.CONNECTED: + return + + self._status = ChannelStatus.CONNECTING + logger.info(f"[NextcloudTalk] Starting channel '{self.channel_id}'") + + server_url = self.config.get("server_url", "").rstrip("/") + if not server_url: + raise ChannelAuthenticationError() + parsed = urlparse(server_url) + if parsed.scheme not in ("https", "http") or not parsed.hostname: + raise ChannelAuthenticationError(f"Invalid server_url format: {server_url}") + bot_user = self.config.get("bot_user", self.config.get("botUser", "")) or os.getenv( + "NEXTCLOUD_TALK_BOT_USER", "" + ) + app_password = self.config.get("app_password", self.config.get("appPassword", "")) or os.getenv( + "NEXTCLOUD_TALK_APP_PASSWORD", + os.getenv("NEXTCLOUD_TALK_API_PASSWORD", os.getenv("NEXTCLOUD_TALK_BOT_SECRET", "")), + ) + if not bot_user or not app_password: + raise ChannelAuthenticationError() + + warnings = collect_security_warnings(self.config) + for w in warnings: + logger.warning(f"[NextcloudTalk] Security warning: {w}") + + network_cfg = self.config.get("network", {}) + proxy_url = network_cfg.get("proxy", "") + self._client = NextcloudTalkClient( + server_url, + bot_user, + app_password, + proxy=proxy_url if proxy_url else None, + bot_secret=app_password, + ) + await self._client.start() + + try: + caps = await probe_capabilities(self._client.session, server_url) + api_version = resolve_api_version(caps) + self._api_base = resolve_api_base(api_version) + talk_version = caps.get("version", "unknown") + logger.info(f"[NextcloudTalk] Talk version: {talk_version}, API: {api_version}") + except Exception as e: + logger.warning(f"[NextcloudTalk] Probe failed: {e}, using default v1 API") + self._api_base = resolve_api_base("v1") + + self._bot_display_name = self.config.get("display_name", "ForcePilot Bot") + self._reaction_level = self.config.get("reaction_level", self.config.get("reactionLevel", "minimal")) + coalesce_key = "block_streaming_coalesce" + self._stream_coalesce = self.config.get(coalesce_key, self.config.get("blockStreamingCoalesce", {})) + + ws = None + if self.config.get("try_websocket", True): + ws = await try_connect_websocket( + server_url, + self._client.basic_auth, + bot_user, + app_password, + self._client.session, + ) + + if ws: + self._monitor_mode = "websocket" + logger.info("[NextcloudTalk] Mode: WebSocket") + else: + self._monitor_mode = "polling" + logger.info("[NextcloudTalk] Mode: Polling (WebSocket unavailable)") + + await self._load_conversations() + logger.info(f"[NextcloudTalk] Loaded {len(self._conversations)} conversations") + + if self._monitor_mode == "websocket": + self._ws_monitor = WebSocketMonitor( + ws=ws, + channel_id=self.channel_id, + basic_auth=self._client.basic_auth, + server_url=server_url, + bot_user=bot_user, + app_password=app_password, + ) + self._ws_monitor.set_on_message(self._handle_message) + self._ws_monitor.set_on_fallback(self._fallback_to_polling) + self._ws_task = asyncio.create_task(self._ws_monitor.run()) + else: + await self._start_polling() + + self._status = ChannelStatus.CONNECTED + logger.info(f"[NextcloudTalk] Channel '{self.channel_id}' started in {self._monitor_mode} mode") + + async def disconnect(self) -> None: + if self._status == ChannelStatus.DISCONNECTED: + return + + logger.info(f"[NextcloudTalk] Stopping channel '{self.channel_id}'") + + if self._ws_monitor: + await self._ws_monitor.stop() + self._ws_monitor = None + + if self._ws_task and not self._ws_task.done(): + self._ws_task.cancel() + try: + await self._ws_task + except asyncio.CancelledError: + pass + self._ws_task = None + + if self._polling_monitor: + await self._polling_monitor.stop() + self._polling_monitor = None + + if self._polling_task and not self._polling_task.done(): + self._polling_task.cancel() + try: + await self._polling_task + except asyncio.CancelledError: + pass + self._polling_task = None + + if self._client: + await self._client.stop() + self._client = None + + self._conversations.clear() + self._status = ChannelStatus.DISCONNECTED + logger.info(f"[NextcloudTalk] Channel '{self.channel_id}' stopped") + + async def send(self, response: ChannelResponse) -> DeliveryResult: + if not self._client or self._status != ChannelStatus.CONNECTED: + return DeliveryResult(success=False, error="Channel not connected") + + _send_start = asyncio.get_event_loop().time() + + payloads = self.format_outbound(response) + token = response.identity.channel_chat_id + + last_result = DeliveryResult(success=False, error="No payloads") + silent = self.config.get("silent", False) + for payload in payloads: + + async def _do_send(p=payload): + return await send_text( + self._client, + self._api_base, + token, + p, + self.config, + self._sent_cache, + silent=silent, + ) + + try: + last_result = await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + self._metrics.record_circuit_breaker_trip() + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + self._metrics.record_send_error() + return DeliveryResult(success=False, error=str(e)) + + _send_elapsed = asyncio.get_event_loop().time() - _send_start + self._metrics.record_send_success(_send_elapsed) + + import time as _time + + self._outbound_count += 1 + self._last_outbound_at = _time.time() + + if self._on_send_callback: + try: + self._on_send_callback(last_result) + except Exception: + pass + + return last_result + + def set_on_send_callback(self, callback: Callable[[DeliveryResult], None] | None) -> None: + self._on_send_callback = callback + + async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + + if isinstance(data, bytes): + media_data = data + else: + return DeliveryResult(success=False, error="Media data must be bytes") + + async def _do_send(): + return await send_media( + self._client, + self._api_base, + chat_id, + media_type, + media_data, + config=self.config, + ) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + + async def _do_send(): + return await send_reaction( + self._client, + self._api_base, + chat_id, + msg_id, + emoji, + config=self.config, + ) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def remove_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + + async def _do_send(): + return await send_reaction_delete( + self._client, + self._api_base, + chat_id, + msg_id, + emoji, + config=self.config, + ) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def get_reactions(self, chat_id: str, msg_id: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + + async def _do_send(): + return await _get_reactions( + self._client, + self._api_base, + chat_id, + msg_id, + config=self.config, + ) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_stream_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + + coalesce = self._stream_coalesce + if coalesce and not finished: + min_chars = coalesce.get("min_chars", 0) + max_delay_ms = coalesce.get("max_delay_ms", 0) + if min_chars > 0 and len(chunk) < min_chars: + self._stream_pending = getattr(self, "_stream_pending", {}) + pending = self._stream_pending.get(chat_id, "") + self._stream_pending[chat_id] = pending + chunk + if max_delay_ms > 0: + await asyncio.sleep(max_delay_ms / 1000) + if finished: + chunk = self._stream_pending.pop(chat_id, pending + chunk) + elif len(self._stream_pending.get(chat_id, "")) < min_chars: + return DeliveryResult(success=True) + else: + chunk = self._stream_pending.pop(chat_id, "") + else: + pending = getattr(self, "_stream_pending", {}).pop(chat_id, "") + if pending: + chunk = pending + chunk + + now = asyncio.get_event_loop().time() + last_edit = self._stream_edit_times.get(chat_id, 0) + if last_edit > 0: + min_interval = 1.0 / self._stream_max_edits_per_sec + elapsed = now - last_edit + if elapsed < min_interval: + await asyncio.sleep(min_interval - elapsed) + self._stream_edit_times[chat_id] = now + + self._stream_msg_ttls[chat_id] = now + + if not msg_id: + identity = self._build_stream_identity(chat_id, msg_id) + response = ChannelResponse( + identity=identity, + content=chunk, + metadata={"reference_id": f"stream-{chat_id}"}, + ) + result = await self.send(response) + if result.success and result.message_id: + self._stream_msg_ids[chat_id] = result.message_id + return result + + async def _do_edit(): + return await _send_edit( + self._client, + self._api_base, + chat_id, + msg_id, + chunk, + self.config, + ) + + try: + return await self._circuit_breaker.call(_do_edit) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def receive(self) -> AsyncIterator[ChannelMessage]: + return + yield # type: ignore[misc] + + def normalize_inbound(self, raw: dict[str, Any]) -> ChannelMessage: + token = raw.get("token", "") + conv = self._conversations.get(token) + room_type = conv.get("type") if conv else None + message = normalize_inbound(raw, self.channel_id, room_type) + message.metadata["provider"] = "nextcloud-talk" + message.metadata["surface"] = "nextcloud-talk" + sender_name = message.metadata.get("nc_actor_display_name", "") + if sender_name: + message.metadata["sender_name"] = sender_name + return message + + def format_outbound(self, response: ChannelResponse) -> list[dict[str, Any]]: + markdown_cfg = self.config.get("markdown", {}) + response_prefix = self.config.get("response_prefix", self.config.get("responsePrefix", "")) + return format_outbound(response, self._bot_display_name, self.text_chunk_limit, markdown_cfg, response_prefix) + + async def edit_message(self, chat_id: str, msg_id: str, content: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + + async def _do_edit(): + return await _send_edit(self._client, self._api_base, chat_id, msg_id, content, self.config) + + try: + return await self._circuit_breaker.call(_do_edit) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def delete_message(self, chat_id: str, msg_id: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + + async def _do_delete(): + return await _send_delete(self._client, self._api_base, chat_id, msg_id, self.config) + + try: + return await self._circuit_breaker.call(_do_delete) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def get_user_info(self, channel_user_id: str) -> dict[str, Any]: + if not self._client: + return {} + try: + result = await self._client.get(f"/ocs/v2.php/cloud/users/{channel_user_id}") + ocs = result.get("ocs", {}) + return ocs.get("data", {}) + except Exception: + return {"id": channel_user_id} + + async def download_media(self, file_id: str) -> bytes: + if not self._client: + raise RuntimeError("Client not initialized") + if not file_id.startswith(self._client.server_url): + raise ValueError("Invalid media URL: must be served by the same Nextcloud instance") + if re.search(r"/\.{1,2}/", file_id): + raise ValueError("Invalid media URL: path traversal detected") + async with self._client.session.get(file_id, headers=self._client.auth_headers()) as resp: + resp.raise_for_status() + return await resp.read() + + def resolve_session_route(self, token: str, default_agent_id: str) -> dict[str, str]: + conv = self._conversations.get(token) + return resolve_session_route(token, conv, default_agent_id) + + async def _handle_message(self, message: ChannelMessage) -> None: + user_id = message.identity.channel_user_id + token = message.identity.channel_chat_id + msg_id = message.identity.channel_message_id + + import time as _time + + self._inbound_count += 1 + self._last_inbound_at = _time.time() + + if msg_id: + if self._dedup_guard.is_invalid(token, msg_id): + pass + elif not self._dedup_guard.claim(token, msg_id): + return + + try: + if message.chat_type == ChatType.DIRECT: + if not check_dm_policy(user_id, self.config): + dm_policy = self.config.get("dm_policy", "pairing") + if dm_policy == "pairing" and self._client: + await send_pairing_challenge( + self._client, + self._api_base, + token, + user_id, + self.config, + ) + logger.debug(f"[NextcloudTalk] DM blocked by policy: {user_id}") + self._dedup_guard.release(token, msg_id) if msg_id else None + return + + if not resolve_dm_enabled(user_id, self.config): + logger.debug(f"[NextcloudTalk] DM disabled by dms config: {user_id}") + self._dedup_guard.release(token, msg_id) if msg_id else None + return + + dm_system_prompt = resolve_dm_system_prompt(user_id, self.config) + if dm_system_prompt: + message.metadata["system_prompt"] = dm_system_prompt + dm_skills = resolve_dm_skills(user_id, self.config) + if dm_skills: + message.metadata["skills_filter"] = dm_skills + else: + if not check_group_policy(token, user_id, self.config): + logger.debug(f"[NextcloudTalk] Group blocked: {user_id} in {token}") + self._dedup_guard.release(token, msg_id) if msg_id else None + return + + if not resolve_room_enabled(token, self.config): + logger.debug(f"[NextcloudTalk] Room disabled: {token}") + self._dedup_guard.release(token, msg_id) if msg_id else None + return + + raw_params = message.metadata.get("nc_message_parameters", {}) + bot_name = self._bot_display_name + if not check_mention_gate( + token, + message.content, + bot_name, + self.config, + raw_params, + ): + logger.debug(f"[NextcloudTalk] Mention gate blocked: {user_id} in {token}") + self._dedup_guard.release(token, msg_id) if msg_id else None + return + + room_system_prompt = resolve_room_system_prompt(token, self.config) + if room_system_prompt: + message.metadata["system_prompt"] = room_system_prompt + room_skills = resolve_room_skills(token, self.config) + if room_skills: + message.metadata["skills_filter"] = room_skills + + conv = self._conversations.get(token) + room_type = conv.get("type") if conv else None + message.metadata["thread_key"] = make_thread_key(self.config.get("agent_id", "default"), token, room_type) + + await super()._handle_message(message) + + if msg_id: + self._dedup_guard.commit(token, msg_id) + except Exception: + if msg_id: + self._dedup_guard.release(token, msg_id) + raise + + async def health_check(self) -> HealthStatus: + if not self._client: + return HealthStatus(status="unhealthy", last_error="Client not initialized") + + self._cleanup_stream_ids() + + try: + caps = await probe_capabilities(self._client.session, self._client.server_url) + spreed_version = caps.get("version", "unknown") if caps else "not detected" + + return HealthStatus( + status="healthy", + metadata={ + "server_url": self._client.server_url, + "talk_version": spreed_version, + "monitor_mode": self._monitor_mode, + "conversations": len(self._conversations), + "dedup_stats": self._dedup_guard.stats(), + "active_stream_count": len(self._stream_msg_ids), + "total_active_streams": len(self._stream_msg_ids), + "activity": self.get_activity_stats(), + "metrics": self._metrics.snapshot(), + }, + last_connected_at=utc_now_naive(), + ) + except Exception as e: + return HealthStatus(status="degraded", last_error=str(e)) + + async def verify_webhook_signature(self, headers: dict, body: bytes) -> bool: + bot_secret = self.config.get("app_password", self.config.get("appPassword", "")) or os.getenv( + "NEXTCLOUD_TALK_APP_PASSWORD", os.getenv("NEXTCLOUD_TALK_BOT_SECRET", "") + ) + if not bot_secret: + logger.warning("[NextcloudTalk] Cannot verify webhook: no bot_secret configured") + return False + signature = headers.get("X-Nextcloud-Talk-SHA256", headers.get("x-nextcloud-talk-sha256", "")) + if not signature: + return False + return verify_hmac_signature(body, signature, bot_secret) + + def _cleanup_stream_ids(self) -> None: + now = asyncio.get_event_loop().time() + ttl = 300 + stale = [cid for cid, ts in self._stream_msg_ttls.items() if now - ts > ttl] + for cid in stale: + self._stream_msg_ids.pop(cid, None) + self._stream_msg_ttls.pop(cid, None) + self._stream_edit_times.pop(cid, None) + if stale: + logger.debug(f"[NextcloudTalk] Cleaned {len(stale)} stale stream IDs") + + async def pre_connect(self) -> dict: + server_url = self.config.get("server_url", "").rstrip("/") + if not server_url: + return {"status": "error", "message": "Missing server_url"} + bot_user = self.config.get("bot_user", "") + app_password = self.config.get("app_password", "") + if not bot_user or not app_password: + return {"status": "error", "message": "Missing bot_user or app_password"} + + client = NextcloudTalkClient(server_url, bot_user, app_password, bot_secret=app_password) + await client.start() + try: + caps = await probe_capabilities(client.session, server_url) + version = caps.get("version", "unknown") if caps else "not detected" + return { + "status": "ok", + "server_url": server_url, + "bot_user": bot_user, + "talk_version": version, + } + except Exception as e: + return {"status": "error", "message": str(e)} + finally: + await client.stop() + + async def _load_conversations(self) -> None: + if not self._client: + return + + path = f"{self._api_base}/conversation" + try: + result = await self._client.get(path) + ocs = result.get("ocs", {}) + data = ocs.get("data", []) + if isinstance(data, list): + for conv in data: + token = conv.get("token", "") + if token: + self._conversations[token] = { + "token": token, + "type": conv.get("type", 2), + "name": conv.get("name", ""), + "display_name": conv.get("displayName", conv.get("name", "")), + } + except Exception as e: + logger.warning(f"[NextcloudTalk] Failed to load conversations: {e}") + + async def create_poll(self, chat_id: str, question: str, options: list[str]) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + payload = {"question": question, "options": options} + + async def _do_create(): + path = f"{self._api_base}/poll/{chat_id}" + try: + result = await self._client.post(path, json=payload) + ocs = result.get("ocs", {}) + if ocs.get("meta", {}).get("status") == "ok": + poll_id = ocs.get("data", {}).get("id", "") + return DeliveryResult(success=True, message_id=poll_id) + return DeliveryResult(success=False, error=ocs.get("meta", {}).get("message", "Unknown error")) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + try: + return await self._circuit_breaker.call(_do_create) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def list_conversations(self) -> list[dict[str, Any]]: + if not self._client: + return [] + path = f"{self._api_base}/conversation" + try: + result = await self._client.get(path) + ocs = result.get("ocs", {}) + data = ocs.get("data", []) + if isinstance(data, list): + return [ + { + "token": conv.get("token", ""), + "type": conv.get("type", 2), + "name": conv.get("name", ""), + "display_name": conv.get("displayName", conv.get("name", "")), + } + for conv in data + if conv.get("token") + ] + return [] + except Exception as e: + logger.warning(f"[NextcloudTalk] Failed to list conversations: {e}") + return [] + + async def _refresh_conversations(self) -> None: + prev_count = len(self._conversations) + await self._load_conversations() + new_count = len(self._conversations) + if prev_count != new_count: + logger.info(f"[NextcloudTalk] Conversations refreshed: {prev_count} 鈫?{new_count}") + if self._polling_monitor: + self._polling_monitor.update_conversations(self._conversations) + + async def _start_polling(self) -> None: + if not self._client: + return + + poll_interval = self.config.get("poll_interval", 0.5) + poll_timeout = self.config.get("poll_timeout", 30) + + self._polling_monitor = PollingMonitor( + client=self._client, + api_base=self._api_base, + channel_id=self.channel_id, + poll_interval=float(poll_interval), + poll_timeout=int(poll_timeout), + config=self.config, + ) + self._polling_monitor.set_on_message(self._handle_message) + self._polling_monitor.set_on_refresh(self._refresh_conversations) + self._polling_task = asyncio.create_task(self._polling_monitor.start(self._conversations)) + + async def _fallback_to_polling(self) -> None: + logger.info("[NextcloudTalk] Falling back to polling mode") + self._monitor_mode = "polling" + self._status = ChannelStatus.CONNECTED + await self._start_polling() + + async def get_account(self, account_id: str) -> dict[str, Any] | None: + session = self._sessions.get(account_id) + if not session: + return None + return { + "account_id": session.account_id, + "label": session.label, + "server_url": session.server_url, + "bot_user": session.bot_user, + "is_primary": account_id == self._primary_account_id, + } + + async def list_accounts(self) -> list[dict[str, Any]]: + return [ + { + "account_id": s.account_id, + "label": s.label, + "server_url": s.server_url, + "bot_user": s.bot_user, + "is_primary": s.account_id == self._primary_account_id, + } + for s in self._sessions.values() + ] + + @property + def primary_account_id(self) -> str | None: + return self._primary_account_id + + @property + def is_multi_account(self) -> bool: + return self._is_multi_account + + def _resolve_accounts(self) -> None: + accounts_cfg = self.config.get("accounts", []) + if isinstance(accounts_cfg, list) and accounts_cfg: + self._is_multi_account = True + for acct in accounts_cfg: + if not isinstance(acct, dict): + continue + aid = acct.get("id", str(len(self._sessions) + 1)) + session = _AccountSession( + account_id=aid, + label=acct.get("label", f"account-{aid}"), + server_url=acct.get("server_url", self.config.get("server_url", "")).rstrip("/"), + bot_user=acct.get("bot_user", self.config.get("bot_user", "")), + app_password=acct.get( + "app_password", + acct.get( + "appPassword", + self.config.get("app_password", self.config.get("appPassword", "")), + ), + ), + config={**self.config, **acct}, + ) + self._sessions[aid] = session + if self._primary_account_id is None: + self._primary_account_id = aid + + if not self._sessions: + self._is_multi_account = False + session = _AccountSession( + account_id="default", + label="default", + server_url=self.config.get("server_url", "").rstrip("/"), + bot_user=self.config.get("bot_user", self.config.get("botUser", "")), + app_password=self.config.get("app_password", self.config.get("appPassword", "")), + config=self.config, + ) + self._sessions["default"] = session + self._primary_account_id = "default" + + if _DEBUG_ACCOUNTS: + for aid, s in self._sessions.items(): + logger.debug( + f"[NextcloudTalk] Account '{aid}': label={s.label}, " + f"server={s.server_url}, bot_user={s.bot_user}, " + f"has_password={'***' if s.app_password else 'no'}" + ) + logger.debug( + f"[NextcloudTalk] Multi-account: {self._is_multi_account}, " + f"primary: {self._primary_account_id}, total: {len(self._sessions)}" + ) + + async def send_voice(self, chat_id: str, audio_data: bytes, mime_type: str = "audio/mp4") -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + return await self.send_media(chat_id, "voice", audio_data) + + async def transcribe_voice(self, chat_id: str, msg_id: str) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + + async def _do_transcribe(): + path = f"{self._api_base}/transcribe/{chat_id}/{msg_id}" + try: + result = await self._client.get(path) + ocs = result.get("ocs", {}) + if ocs.get("meta", {}).get("status") == "ok": + transcription = ocs.get("data", {}).get("transcription", ocs.get("data", {}).get("text", "")) + return DeliveryResult(success=True, metadata={"transcription": transcription}) + return DeliveryResult(success=False, error="Transcription not available") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + try: + return await self._circuit_breaker.call(_do_transcribe) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def analyze_image(self, chat_id: str, msg_id: str, prompt: str = "") -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + + async def _do_analyze(): + base_path = f"{self._api_base}/analyze/{chat_id}/{msg_id}" + path = base_path + try: + if prompt: + result = await self._client.post(path, json={"prompt": prompt}) + else: + result = await self._client.get(path) + ocs = result.get("ocs", {}) + if ocs.get("meta", {}).get("status") == "ok": + analysis = ocs.get("data", {}).get("analysis", ocs.get("data", {}).get("result", "")) + return DeliveryResult(success=True, metadata={"analysis": analysis}) + return DeliveryResult(success=False, error="Image analysis not available") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + try: + return await self._circuit_breaker.call(_do_analyze) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def request_approval( + self, + chat_id: str, + action: str, + details: dict[str, Any] | None = None, + ) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Channel not connected") + payload = {"action": action, "details": details or {}} + + async def _do_approve(): + path = f"{self._api_base}/approval/{chat_id}" + try: + result = await self._client.post(path, json=payload) + ocs = result.get("ocs", {}) + if ocs.get("meta", {}).get("status") == "ok": + approval_id = ocs.get("data", {}).get("id", "") + return DeliveryResult(success=True, message_id=approval_id) + return DeliveryResult( + success=False, + error=ocs.get("meta", {}).get("message", "Approval request failed"), + ) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + try: + return await self._circuit_breaker.call(_do_approve) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def reload_config(self, new_config: dict[str, Any]) -> None: + logger.info("[NextcloudTalk] Reloading config, will disconnect and reconnect") + await self.disconnect() + self.config = new_config + await self.connect() + + def get_activity_stats(self) -> dict[str, Any]: + import time as _time + + now = _time.time() + return { + "inbound_count": self._inbound_count, + "outbound_count": self._outbound_count, + "last_inbound_seconds_ago": round(now - self._last_inbound_at, 1) if self._last_inbound_at else None, + "last_outbound_seconds_ago": round(now - self._last_outbound_at, 1) if self._last_outbound_at else None, + } diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/approval_auth.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/approval_auth.py new file mode 100644 index 00000000..8de62492 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/approval_auth.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import logging +from typing import Any + +from .normalize import normalize_user_target, strip_nc_prefix + +logger = logging.getLogger(__name__) + + +def resolve_approver_id(user_id: str) -> str: + approver_id = strip_nc_prefix(user_id) + return f"nc:{approver_id}" + + +def resolve_allow_from_approvers(config: dict[str, Any]) -> list[str]: + allow_from = config.get("allow_from", config.get("allowFrom", [])) + return [normalize_user_target(strip_nc_prefix(entry)) for entry in allow_from] + + +def check_approval_required( + user_id: str, + config: dict[str, Any], + paired_users: set[str] | None = None, +) -> dict[str, Any]: + dm_policy = config.get("dm_policy", "pairing") + + if dm_policy == "open": + return {"approval_required": False, "reason": "dm_policy is open"} + + if dm_policy == "disabled": + return {"approval_required": True, "reason": "dm_policy is disabled", "action": "block"} + + if dm_policy == "allowlist": + allow_from = config.get("allow_from", config.get("allowFrom", [])) + approved = any( + normalize_user_target(strip_nc_prefix(e)) == normalize_user_target(strip_nc_prefix(user_id)) + for e in allow_from + ) + if approved: + return {"approval_required": False, "reason": "User is in allowlist"} + return {"approval_required": True, "reason": "User not in allowlist", "action": "block"} + + if dm_policy == "pairing": + if paired_users and user_id in paired_users: + return {"approval_required": False, "reason": "User is already paired"} + return { + "approval_required": True, + "reason": "User not paired", + "action": "pair", + "approver_id": resolve_approver_id(user_id), + } + + return {"approval_required": True, "reason": f"Unknown dm_policy: {dm_policy}"} diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/chunking.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/chunking.py new file mode 100644 index 00000000..c440b581 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/chunking.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + + +def chunk_text(text: str, limit: int = 4000, mode: str = "markdown") -> list[str]: + if len(text) <= limit: + return [text] + + if mode == "markdown": + return _chunk_markdown(text, limit) + elif mode == "newline": + return _chunk_by_newline(text, limit) + return _chunk_by_length(text, limit) + + +def _chunk_markdown(text: str, limit: int) -> list[str]: + chunks: list[str] = [] + remainder = text + + while len(remainder) > limit: + cut = remainder.rfind("\n\n", 0, limit) + if cut < limit // 2: + cut = remainder.rfind("\n", 0, limit) + if cut < limit // 4: + cut = remainder.rfind(". ", 0, limit) + if cut < 0: + cut = limit + + chunk = remainder[: cut + 1].rstrip() + chunks.append(chunk) + remainder = remainder[cut + 1 :].lstrip() + + if remainder.strip(): + chunks.append(remainder.strip()) + + return chunks + + +def _chunk_by_newline(text: str, limit: int) -> list[str]: + chunks: list[str] = [] + current = "" + for line in text.split("\n"): + if len(current) + len(line) + 1 > limit and current: + chunks.append(current.rstrip()) + current = line + else: + current = f"{current}\n{line}".strip() + if current: + chunks.append(current) + return chunks + + +def _chunk_by_length(text: str, limit: int) -> list[str]: + chunks: list[str] = [] + for i in range(0, len(text), limit): + chunks.append(text[i : i + limit]) + return chunks diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/client.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/client.py new file mode 100644 index 00000000..69f4455e --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/client.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import logging +from typing import Any + +import aiohttp + +from yuxi.channels.exceptions import ChannelNotConnectedError + +logger = logging.getLogger(__name__) + + +def _compute_hmac_signature(body: bytes, secret: str) -> str: + return base64.b64encode(hmac.digest(secret.encode(), body, hashlib.sha256)).decode() + + +def verify_hmac_signature(body: bytes, signature_header: str, secret: str) -> bool: + if not signature_header or not secret: + return False + expected = _compute_hmac_signature(body, secret) + return hmac.compare_digest(expected, signature_header) + + +class NextcloudTalkClient: + def __init__( + self, + server_url: str, + bot_user: str, + app_password: str, + proxy: str | None = None, + bot_secret: str | None = None, + ): + self.server_url = server_url.rstrip("/") + self.bot_user = bot_user + self.basic_auth = base64.b64encode(f"{bot_user}:{app_password}".encode()).decode() + self.proxy = proxy + self.bot_secret = bot_secret + self._session: aiohttp.ClientSession | None = None + + async def start(self) -> None: + if self._session is None: + connector_kwargs: dict[str, Any] = {"limit": 10, "ttl_dns_cache": 300} + timeout = aiohttp.ClientTimeout(total=60) + session_kwargs: dict[str, Any] = { + "connector": aiohttp.TCPConnector(**connector_kwargs), + "timeout": timeout, + "headers": {"User-Agent": "ForcePilot-NextcloudTalk-Adapter/1.0"}, + } + if self.proxy: + session_kwargs["proxy"] = self.proxy + self._session = aiohttp.ClientSession(**session_kwargs) + + async def stop(self) -> None: + if self._session: + await self._session.close() + self._session = None + + @property + def session(self) -> aiohttp.ClientSession: + if self._session is None: + raise ChannelNotConnectedError() + return self._session + + def auth_headers(self) -> dict[str, str]: + return {"Authorization": f"Basic {self.basic_auth}"} + + def api_url(self, path: str) -> str: + return f"{self.server_url}{path}" + + def _sign_body(self, body_bytes: bytes) -> dict[str, str]: + if not self.bot_secret: + return {} + return {"X-Nextcloud-Talk-SHA256": _compute_hmac_signature(body_bytes, self.bot_secret)} + + async def get(self, path: str, **kwargs) -> dict[str, Any]: + url = self.api_url(path) + headers = {**self.auth_headers(), "OCS-APIRequest": "true", "Accept": "application/json"} + headers.update(kwargs.pop("headers", {})) + async with self.session.get(url, headers=headers, **kwargs) as resp: + resp.raise_for_status() + return await resp.json() + + async def post(self, path: str, json_data: dict[str, Any] | None = None, **kwargs) -> dict[str, Any]: + url = self.api_url(path) + headers = {**self.auth_headers(), "OCS-APIRequest": "true", "Accept": "application/json"} + body_bytes = json.dumps(json_data).encode() if json_data else b"" + headers.update(self._sign_body(body_bytes)) + headers.update(kwargs.pop("headers", {})) + async with self.session.post(url, headers=headers, json=json_data, **kwargs) as resp: + resp.raise_for_status() + return await resp.json() + + async def post_form(self, path: str, form: aiohttp.FormData, **kwargs) -> dict[str, Any]: + url = self.api_url(path) + headers = {**self.auth_headers(), "OCS-APIRequest": "true", "Accept": "application/json"} + headers.update(kwargs.pop("headers", {})) + async with self.session.post(url, headers=headers, data=form, **kwargs) as resp: + resp.raise_for_status() + return await resp.json() + + async def put(self, path: str, json_data: dict[str, Any] | None = None, **kwargs) -> dict[str, Any]: + url = self.api_url(path) + headers = {**self.auth_headers(), "OCS-APIRequest": "true", "Accept": "application/json"} + body_bytes = json.dumps(json_data).encode() if json_data else b"" + headers.update(self._sign_body(body_bytes)) + headers.update(kwargs.pop("headers", {})) + async with self.session.put(url, headers=headers, json=json_data, **kwargs) as resp: + resp.raise_for_status() + return await resp.json() + + async def delete(self, path: str, **kwargs) -> dict[str, Any]: + url = self.api_url(path) + headers = {**self.auth_headers(), "OCS-APIRequest": "true", "Accept": "application/json"} + headers.update(kwargs.pop("headers", {})) + async with self.session.delete(url, headers=headers, **kwargs) as resp: + resp.raise_for_status() + return await resp.json() diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/config.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/config.py new file mode 100644 index 00000000..9aaecdc6 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/config.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +from typing import Any, Literal + +MatchSource = Literal["direct", "wildcard", "none"] + + +def resolve_room_config( + token: str, + config: dict[str, Any], +) -> dict[str, Any]: + rooms_cfg = config.get("rooms", {}) + + if not rooms_cfg: + return {"_match_source": "none"} + + match_source: MatchSource = "wildcard" + merged: dict[str, Any] = dict(rooms_cfg.get("*", {})) + + if rooms_cfg.get("*"): + match_source = "wildcard" + + if token in rooms_cfg: + merged.update(rooms_cfg[token]) + match_source = "direct" + + merged["_match_source"] = match_source + return merged + + +def resolve_room_enabled(token: str, config: dict[str, Any]) -> bool: + room_cfg = resolve_room_config(token, config) + return room_cfg.get("enabled", room_cfg.get("enabled_", True)) + + +def resolve_room_require_mention(token: str, config: dict[str, Any]) -> bool: + room_cfg = resolve_room_config(token, config) + return room_cfg.get("requireMention", room_cfg.get("require_mention", True)) + + +def resolve_room_system_prompt(token: str, config: dict[str, Any]) -> str | None: + room_cfg = resolve_room_config(token, config) + return room_cfg.get("systemPrompt", room_cfg.get("system_prompt")) + + +def resolve_room_allow_from(token: str, config: dict[str, Any]) -> list[str]: + room_cfg = resolve_room_config(token, config) + return room_cfg.get("allowFrom", room_cfg.get("allow_from", [])) + + +def resolve_room_skills(token: str, config: dict[str, Any]) -> list[str]: + room_cfg = resolve_room_config(token, config) + return room_cfg.get("skills", []) + + +def resolve_dm_config(user_id: str, config: dict[str, Any]) -> dict[str, Any]: + dms_cfg = config.get("dms", {}) + if not dms_cfg: + return {} + merged: dict[str, Any] = dict(dms_cfg.get("*", {})) + if user_id in dms_cfg: + merged.update(dms_cfg[user_id]) + return merged + + +def resolve_dm_enabled(user_id: str, config: dict[str, Any]) -> bool: + dm_cfg = resolve_dm_config(user_id, config) + return dm_cfg.get("enabled", True) + + +def resolve_dm_system_prompt(user_id: str, config: dict[str, Any]) -> str | None: + dm_cfg = resolve_dm_config(user_id, config) + return dm_cfg.get("systemPrompt", dm_cfg.get("system_prompt")) + + +def resolve_dm_skills(user_id: str, config: dict[str, Any]) -> list[str]: + dm_cfg = resolve_dm_config(user_id, config) + return dm_cfg.get("skills", []) + + +def resolve_history_limit(token: str, config: dict[str, Any], chat_type: str = "group") -> int: + if chat_type == "direct": + dm_limit = resolve_dm_config(token, config).get("historyLimit") + if dm_limit is not None: + return int(dm_limit) + return config.get("dmHistoryLimit", config.get("dm_history_limit", 0)) + room_cfg = resolve_room_config(token, config) + room_limit = room_cfg.get("historyLimit") + if room_limit is not None: + return int(room_limit) + return config.get("historyLimit", config.get("history_limit", 0)) diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/config_schema.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/config_schema.py new file mode 100644 index 00000000..a2613092 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/config_schema.py @@ -0,0 +1,188 @@ +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, Field, model_validator + + +class NextcloudTalkRoomConfig(BaseModel): + enabled: bool = Field(default=True, description="Enable bot responses in this room") + system_prompt: str | None = Field( + default=None, alias="systemPrompt", description="Override system prompt for this room" + ) + require_mention: bool = Field( + default=True, alias="requireMention", description="Only respond when bot is @mentioned" + ) + allow_from: list[str] = Field( + default_factory=list, alias="allowFrom", description="Allowlist of user IDs for this room" + ) + skills: list[str] = Field(default_factory=list, description="Skills filter for this room") + history_limit: int = Field( + default=0, alias="historyLimit", description="Max history messages to load (0=unlimited)" + ) + + model_config = {"populate_by_name": True, "extra": "allow"} + + +class NextcloudTalkDMConfig(BaseModel): + enabled: bool = Field(default=True, description="Enable bot responses in this DM") + system_prompt: str | None = Field( + default=None, alias="systemPrompt", description="Override system prompt for this DM" + ) + allow_from: list[str] = Field( + default_factory=list, alias="allowFrom", description="Allowlist of user IDs for this DM" + ) + skills: list[str] = Field(default_factory=list, description="Skills filter for this DM") + history_limit: int = Field( + default=0, alias="historyLimit", description="Max history messages to load (0=unlimited)" + ) + + model_config = {"populate_by_name": True, "extra": "allow"} + + +class NextcloudTalkWebhookConfig(BaseModel): + enabled: bool = Field(default=False, description="Enable standalone webhook server") + port: int = Field(default=8788, description="Webhook server listen port") + host: str = Field(default="0.0.0.0", description="Webhook server bind address") + path: str = Field(default="/nextcloud-talk-webhook", description="Webhook endpoint path") + rate_limit_max_requests: int = Field( + default=10, + alias="rateLimitMaxRequests", + description="Max failed auth attempts in window", + ) + rate_limit_window_s: int = Field( + default=60, + alias="rateLimitWindowS", + description="Rate limiting window in seconds", + ) + rate_limit_lockout_s: int = Field( + default=300, + alias="rateLimitLockoutS", + description="Lockout duration after exceeding rate limit", + ) + max_body_size: int = Field( + default=1 * 1024 * 1024, + alias="maxBodySize", + description="Maximum webhook request body size in bytes (default: 1MB)", + ) + pre_auth_max_body_size: int = Field( + default=64 * 1024, + alias="preAuthMaxBodySize", + description="Maximum body size before HMAC auth verification (default: 64KB)", + ) + + model_config = {"populate_by_name": True, "extra": "allow"} + + +class NextcloudTalkAccountConfig(BaseModel): + server_url: str = Field(default="", alias="serverUrl", description="Nextcloud server base URL (https://...)") + bot_user: str = Field(default="", alias="botUser", description="Bot username for authentication") + app_password: str = Field( + default="", + alias="appPassword", + description="App password or device-specific password", + ) + display_name: str = Field( + default="ForcePilot Bot", + alias="displayName", + description="Bot display name shown in chats", + ) + dm_policy: str = Field( + default="pairing", + alias="dmPolicy", + description="DM policy: open|pairing|allowlist|disabled", + ) + group_policy: str = Field( + default="allowlist", + alias="groupPolicy", + description="Group chat policy: open|allowlist|disabled", + ) + try_websocket: bool = Field(default=True, alias="tryWebsocket", description="Attempt WebSocket monitor mode") + reaction_level: str = Field( + default="minimal", + alias="reactionLevel", + description="Reaction verbosity: minimal|normal|verbose", + ) + poll_interval: float = Field(default=0.5, description="Polling interval in seconds") + poll_timeout: int = Field(default=30, description="Polling long-poll timeout in seconds") + allow_from: list[str] = Field(default_factory=list, alias="allowFrom", description="Global allowlist of user IDs") + group_allow_from: list[str] = Field( + default_factory=list, alias="groupAllowFrom", description="Global group allowlist" + ) + rooms: dict[str, NextcloudTalkRoomConfig] = Field( + default_factory=dict, + description="Per-room configuration overrides, keyed by room token", + ) + dms: dict[str, NextcloudTalkDMConfig] = Field( + default_factory=dict, + description="Per-user DM config overrides, keyed by user ID", + ) + groups: dict[str, Any] = Field(default_factory=dict, description="Group policy details") + accounts: dict[str, Any] = Field(default_factory=dict, alias="accounts", description="Multi-account configurations") + retry: dict[str, int] = Field( + default_factory=lambda: {"attempts": 3, "min_delay_ms": 400, "max_delay_ms": 30000}, + description="Send retry policy: attempts, min/max delay", + ) + history_limit: int = Field(default=0, alias="historyLimit", description="Global history limit for group chats") + dm_history_limit: int = Field(default=0, alias="dmHistoryLimit", description="Global history limit for DMs") + markdown: dict[str, Any] = Field(default_factory=dict, description="Markdown rendering options") + block_streaming_coalesce: dict[str, Any] = Field( + default_factory=dict, + alias="blockStreamingCoalesce", + description="Block streaming coalesce config (min_chars, max_delay_ms)", + ) + response_prefix: str = Field( + default="", + alias="responsePrefix", + description="Text prepended to every bot response", + ) + silent: bool = Field(default=False, description="Send messages with silent flag") + chunk_mode: str = Field( + default="split", + alias="chunkMode", + description="Chunking strategy: split|stop|truncate", + ) + media_max_mb: int = Field(default=20, alias="mediaMaxMb", description="Max media file size in MB") + webhook: NextcloudTalkWebhookConfig = Field( + default_factory=NextcloudTalkWebhookConfig, + description="Standalone webhook server configuration", + ) + dangerously_allow_private_network: bool = Field( + default=False, + alias="dangerouslyAllowPrivateNetwork", + description="Allow connections to private/reserved IPs (SSRF risk)", + ) + network: dict[str, Any] = Field(default_factory=dict, description="Network config: proxy, etc.") + + model_config = {"populate_by_name": True, "extra": "allow"} + + @model_validator(mode="after") + def validate_server_url(self) -> NextcloudTalkAccountConfig: + if self.server_url and not (self.server_url.startswith("https://") or self.server_url.startswith("http://")): + raise ValueError(f"server_url must start with https:// or http://, got: {self.server_url}") + return self + + @model_validator(mode="after") + def validate_dm_policy(self) -> NextcloudTalkAccountConfig: + valid = {"open", "pairing", "allowlist", "disabled"} + if self.dm_policy not in valid: + raise ValueError(f"dm_policy must be one of {sorted(valid)}, got: {self.dm_policy}") + return self + + @model_validator(mode="after") + def validate_group_policy(self) -> NextcloudTalkAccountConfig: + valid = {"open", "allowlist", "disabled"} + if self.group_policy not in valid: + raise ValueError(f"group_policy must be one of {sorted(valid)}, got: {self.group_policy}") + return self + + @model_validator(mode="after") + def validate_chunk_mode(self) -> NextcloudTalkAccountConfig: + valid = {"split", "stop", "truncate"} + if self.chunk_mode not in valid: + raise ValueError(f"chunk_mode must be one of {sorted(valid)}, got: {self.chunk_mode}") + return self + + +def validate_config(config: dict[str, Any]) -> NextcloudTalkAccountConfig: + return NextcloudTalkAccountConfig.model_validate(config) diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/credentials.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/credentials.py new file mode 100644 index 00000000..947ce2a2 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/credentials.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import logging +import os + +logger = logging.getLogger(__name__) + + +def read_secret_file(filepath: str) -> str: + if not filepath: + raise ValueError("filepath cannot be empty") + + real_path = os.path.realpath(filepath) + if real_path != os.path.abspath(filepath): + raise ValueError("Secret file path contains symlinks, which is not allowed") + + try: + with open(filepath, encoding="utf-8") as f: + content = f.read() + except FileNotFoundError: + raise ValueError(f"Secret file not found: {filepath}") + except PermissionError: + raise ValueError(f"Permission denied reading secret file: {filepath}") + + return content.strip() + + +def resolve_credential( + direct_value: str | None, + file_path: str | None, + env_var: str | None = None, +) -> str: + if direct_value: + return direct_value + if file_path: + return read_secret_file(file_path) + if env_var: + val = os.getenv(env_var, "") + if val: + return val + return "" + + +def resolve_credential_with_source( + direct_value: str | None, + file_path: str | None, + env_var: str | None = None, +) -> tuple[str, str]: + if direct_value: + return direct_value, "config" + if file_path: + return read_secret_file(file_path), "secretFile" + if env_var: + val = os.getenv(env_var, "") + if val: + return val, "env" + return "", "none" diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/dedup.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/dedup.py new file mode 100644 index 00000000..735cd2bf --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/dedup.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import json +import logging +import os +import time +from collections import OrderedDict + +logger = logging.getLogger(__name__) + +_DEDUP_TTL = 86400 +_MAX_ENTRIES = 10000 +_PERSIST_PATH = os.getenv( + "NEXTCLOUDTALK_DEDUP_FILE", + os.path.join(os.path.dirname(__file__), ".dedup_cache.json"), +) + + +class NextcloudTalkDedupGuard: + def __init__(self, ttl: int = _DEDUP_TTL, max_entries: int = _MAX_ENTRIES): + self._ttl = ttl + self._max_entries = max_entries + self._pending: dict[str, float] = {} + self._committed: OrderedDict[str, float] = OrderedDict() + self._invalid: set[str] = set() + self._loaded = False + + def _load_persisted(self) -> None: + if self._loaded: + return + self._loaded = True + try: + if os.path.exists(_PERSIST_PATH): + with open(_PERSIST_PATH, encoding="utf-8") as f: + data = json.load(f) + now = time.time() + count = 0 + for key, ts in data.items(): + if now - ts < self._ttl: + self._committed[key] = ts + count += 1 + if count: + logger.debug(f"[NextcloudTalk] Dedup: loaded {count} entries from persist") + except Exception as e: + logger.warning(f"[NextcloudTalk] Dedup: failed to load persist: {e}") + + def _save_persisted(self) -> None: + try: + data = dict(self._committed) + with open(_PERSIST_PATH, "w", encoding="utf-8") as f: + json.dump(data, f) + except Exception as e: + logger.warning(f"[NextcloudTalk] Dedup: failed to persist: {e}") + + def _make_key(self, token: str, message_id: str) -> str: + return f"{token}:{message_id}" + + def _gc(self) -> None: + now = time.time() + stale = [k for k, ts in self._committed.items() if now - ts > self._ttl] + for k in stale: + del self._committed[k] + while len(self._committed) > self._max_entries: + self._committed.popitem(last=False) + + def claim(self, token: str, message_id: str) -> bool: + if not token or not message_id: + return True + self._load_persisted() + key = self._make_key(token, message_id) + if key in self._committed: + logger.debug(f"[NextcloudTalk] Dedup: duplicate claim rejected for {key}") + return False + if key in self._pending: + logger.debug(f"[NextcloudTalk] Dedup: already pending {key}") + return False + self._pending[key] = time.time() + return True + + def commit(self, token: str, message_id: str) -> None: + if not token or not message_id: + return + key = self._make_key(token, message_id) + self._pending.pop(key, None) + self._committed[key] = time.time() + self._gc() + + def release(self, token: str, message_id: str) -> None: + if not token or not message_id: + return + key = self._make_key(token, message_id) + self._pending.pop(key, None) + + def mark_invalid(self, token: str, message_id: str) -> None: + if not token or not message_id: + return + key = self._make_key(token, message_id) + self._pending.pop(key, None) + self._invalid.add(key) + + def is_invalid(self, token: str, message_id: str) -> bool: + if not token or not message_id: + return False + key = self._make_key(token, message_id) + return key in self._invalid + + def is_duplicate(self, token: str, message_id: str) -> bool: + if not token or not message_id: + return False + self._load_persisted() + key = self._make_key(token, message_id) + if key in self._committed: + return True + self._gc() + return key in self._committed + + def stats(self) -> dict[str, int]: + self._gc() + return { + "committed": len(self._committed), + "pending": len(self._pending), + "invalid": len(self._invalid), + } + + def flush(self) -> None: + self._gc() + self._save_persisted() diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/directory.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/directory.py new file mode 100644 index 00000000..7b372d9f --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/directory.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import logging +from typing import Any + +from yuxi.channels.models import ChannelType, ChatType + +from .client import NextcloudTalkClient + +logger = logging.getLogger(__name__) + + +async def list_peers( + client: NextcloudTalkClient, + api_base: str, + channel_id: str = "nextcloud-talk", +) -> list[dict[str, Any]]: + try: + path = f"{api_base}/conversation" + result = await client.get(path) + ocs = result.get("ocs", {}) + conversations = ocs.get("data", []) + + peers: list[dict[str, Any]] = [] + if not isinstance(conversations, list): + return peers + + for conv in conversations: + conv_type = conv.get("type", 2) + if conv_type not in (0, 1): + continue + token = conv.get("token", "") + peers.append( + { + "id": f"nextcloud-talk:{token}", + "channel_id": channel_id, + "channel_type": ChannelType.NEXTCLOUDTALK, + "chat_type": ChatType.DIRECT if conv_type == 1 else ChatType.GROUP, + "token": token, + "name": conv.get("name", ""), + "display_name": conv.get("displayName", conv.get("name", "")), + "conversation_type": conv_type, + } + ) + + return peers + except Exception as e: + logger.warning(f"[NextcloudTalk] Failed to list peers: {e}") + return [] + + +async def list_groups( + client: NextcloudTalkClient, + api_base: str, + channel_id: str = "nextcloud-talk", +) -> list[dict[str, Any]]: + try: + path = f"{api_base}/conversation" + result = await client.get(path) + ocs = result.get("ocs", {}) + conversations = ocs.get("data", []) + + groups: list[dict[str, Any]] = [] + if not isinstance(conversations, list): + return groups + + for conv in conversations: + conv_type = conv.get("type", 2) + if conv_type not in (2, 3): + continue + token = conv.get("token", "") + groups.append( + { + "id": f"nextcloud-talk:room:{token}", + "channel_id": channel_id, + "channel_type": ChannelType.NEXTCLOUDTALK, + "chat_type": ChatType.GROUP, + "token": token, + "name": conv.get("name", ""), + "display_name": conv.get("displayName", conv.get("name", "")), + "conversation_type": conv_type, + } + ) + + return groups + except Exception as e: + logger.warning(f"[NextcloudTalk] Failed to list groups: {e}") + return [] + + +async def search_peers( + client: NextcloudTalkClient, + api_base: str, + query: str, + channel_id: str = "nextcloud-talk", +) -> list[dict[str, Any]]: + query_lower = query.lower() + all_peers = await list_peers(client, api_base, channel_id) + return [ + p + for p in all_peers + if query_lower in p.get("name", "").lower() or query_lower in p.get("display_name", "").lower() + ] diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/doctor.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/doctor.py new file mode 100644 index 00000000..a27fc9b2 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/doctor.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +import logging +import os +from typing import Any + +from .probe import probe_capabilities, resolve_api_version +from .security import collect_security_warnings + +logger = logging.getLogger(__name__) + + +async def run_diagnostics( + client, + config: dict[str, Any], + adapter=None, +) -> list[dict[str, Any]]: + results: list[dict[str, Any]] = [] + + results.append(_check_credentials(config)) + results.append(_check_server_url(config)) + results.append(_check_server_url_ssl(config)) + results.append(_check_environment_variables()) + + if client and client._session: + results.append(await _check_api_connectivity(client, config)) + + results.extend(_check_security_policies(config)) + results.append(_check_room_configs(config)) + results.append(_check_dm_configs(config)) + results.append(_check_multi_account(config)) + results.append(_check_webhook_config(config)) + + if adapter: + results.append(_check_runtime_status(adapter)) + + return results + + +def _check_credentials(config: dict[str, Any]) -> dict[str, Any]: + bot_user = config.get("bot_user", config.get("botUser", "")) + app_password = config.get("app_password", config.get("appPassword", "")) + if not bot_user or not app_password: + return {"check": "credentials", "status": "fail", "detail": "bot_user or app_password is missing"} + return {"check": "credentials", "status": "pass", "detail": f"Bot user: {bot_user}"} + + +def _check_server_url(config: dict[str, Any]) -> dict[str, Any]: + server_url = config.get("server_url", "").rstrip("/") + if not server_url: + return {"check": "server_url", "status": "fail", "detail": "server_url is missing"} + if not (server_url.startswith("https://") or server_url.startswith("http://")): + return {"check": "server_url", "status": "fail", "detail": f"Invalid URL format: {server_url}"} + return {"check": "server_url", "status": "pass", "detail": server_url} + + +async def _check_api_connectivity(client, config: dict[str, Any]) -> dict[str, Any]: + try: + server_url = client.server_url + caps = await probe_capabilities(client.session, server_url) + if caps: + version = caps.get("version", "unknown") + api_version = resolve_api_version(caps) + return { + "check": "api_connectivity", + "status": "pass", + "detail": f"Talk version: {version}, API: {api_version}", + } + return {"check": "api_connectivity", "status": "fail", "detail": "No Talk capabilities found"} + except Exception as e: + return {"check": "api_connectivity", "status": "fail", "detail": str(e)} + + +def _check_security_policies(config: dict[str, Any]) -> list[dict[str, Any]]: + warnings = collect_security_warnings(config) + if warnings: + return [ + { + "check": "security_policies", + "status": "warn", + "detail": f"{len(warnings)} warning(s)", + "warnings": warnings, + } + ] + return [{"check": "security_policies", "status": "pass", "detail": "No warnings"}] + + +def _check_room_configs(config: dict[str, Any]) -> dict[str, Any]: + rooms = config.get("rooms", {}) + if not rooms: + return {"check": "room_configs", "status": "info", "detail": "No per-room configuration"} + disabled_rooms = [t for t, c in rooms.items() if not c.get("enabled", c.get("enabled_", True))] + return { + "check": "room_configs", + "status": "info", + "detail": f"{len(rooms)} room(s) configured, {len(disabled_rooms)} disabled", + } + + +def _check_dm_configs(config: dict[str, Any]) -> dict[str, Any]: + dms = config.get("dms", {}) + if not dms: + return {"check": "dm_configs", "status": "info", "detail": "No per-DM configuration"} + disabled_dms = [u for u, c in dms.items() if not c.get("enabled", True)] + return { + "check": "dm_configs", + "status": "info", + "detail": f"{len(dms)} DM config(s), {len(disabled_dms)} disabled", + } + + +def _check_server_url_ssl(config: dict[str, Any]) -> dict[str, Any]: + server_url = config.get("server_url", "").rstrip("/") + if server_url.startswith("https://"): + return {"check": "server_url_ssl", "status": "pass", "detail": "HTTPS enabled"} + if server_url.startswith("http://"): + return {"check": "server_url_ssl", "status": "warn", "detail": "HTTP (non-SSL) in use"} + return {"check": "server_url_ssl", "status": "info", "detail": "No server URL configured"} + + +def _check_environment_variables() -> dict[str, Any]: + found: list[str] = [] + missing: list[str] = [] + + for var in ( + "NEXTCLOUD_TALK_BOT_USER", + "NEXTCLOUD_TALK_BOT_SECRET", + "NEXTCLOUD_TALK_APP_PASSWORD", + "NEXTCLOUD_TALK_API_PASSWORD", + "OPENCLAW_DEBUG_NEXTCLOUD_TALK_ACCOUNTS", + ): + if os.getenv(var): + found.append(var) + else: + missing.append(var) + + return { + "check": "environment_variables", + "status": "info", + "detail": f"{len(found)} set, {len(missing)} not set", + "set": found, + "not_set": missing, + } + + +def _check_multi_account(config: dict[str, Any]) -> dict[str, Any]: + accounts = config.get("accounts", []) + if isinstance(accounts, list) and accounts: + return { + "check": "multi_account", + "status": "info", + "detail": f"{len(accounts)} accounts configured", + "accounts": [ + {"id": a.get("id", "?"), "label": a.get("label", "?")} + for a in accounts if isinstance(a, dict) + ], + } + return {"check": "multi_account", "status": "info", "detail": "Single account mode (default)"} + + +def _check_webhook_config(config: dict[str, Any]) -> dict[str, Any]: + webhook = config.get("webhook", {}) + if not webhook: + return {"check": "webhook", "status": "info", "detail": "Webhook server not configured"} + enabled = webhook.get("enabled", False) + if enabled: + return { + "check": "webhook", + "status": "info", + "detail": f"Enabled (port={webhook.get('port', '?')}, path={webhook.get('path', '?')})", + } + return {"check": "webhook", "status": "info", "detail": "Webhook server disabled"} + + +def _check_runtime_status(adapter) -> dict[str, Any]: + try: + stats = adapter.get_activity_stats() + metrics = adapter._metrics.snapshot() if hasattr(adapter, "_metrics") else {} + return { + "check": "runtime_status", + "status": "pass", + "detail": f"Connected: {adapter.status == 'connected'}", + "activity": stats, + "metrics": metrics, + } + except Exception as e: + return {"check": "runtime_status", "status": "warn", "detail": str(e)} diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/formatter.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/formatter.py new file mode 100644 index 00000000..ec3cafa6 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/formatter.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import re +import uuid +from typing import Any + +from yuxi.channels.models import ChannelResponse, MessageType + +from .chunking import chunk_text + + +def convert_markdown_tables(text: str) -> str: + lines = text.split("\n") + result: list[str] = [] + i = 0 + while i < len(lines): + line = lines[i] + if re.match(r"^\|.*\|$", line.strip()) and i + 1 < len(lines): + next_line = lines[i + 1] + if re.match(r"^\|[\s\-:|]+\|$", next_line.strip()): + header = _parse_table_row(line) + rows: list[list[str]] = [] + i += 2 + while i < len(lines) and re.match(r"^\|.*\|$", lines[i].strip()): + rows.append(_parse_table_row(lines[i])) + i += 1 + result.extend(_table_to_text(header, rows)) + continue + result.append(line) + i += 1 + return "\n".join(result) + + +def _parse_table_row(line: str) -> list[str]: + return [cell.strip() for cell in line.strip().strip("|").split("|")] + + +def _table_to_text(header: list[str], rows: list[list[str]]) -> list[str]: + cols = len(header) + col_widths = [len(h) for h in header] + for row in rows: + for c in range(min(cols, len(row))): + col_widths[c] = max(col_widths[c], len(row[c])) + header_line = " | ".join(h.ljust(col_widths[i]) for i, h in enumerate(header)) + sep_line = "-+-".join("-" * col_widths[i] for i in range(cols)) + result = [header_line, sep_line] + for row in rows: + result.append(" | ".join((row[c] if c < len(row) else "").ljust(col_widths[c]) for c in range(cols))) + return result + + +def format_outbound( + response: ChannelResponse, + bot_display_name: str, + text_chunk_limit: int = 4000, + markdown_config: dict[str, Any] | None = None, + response_prefix: str = "", +) -> list[dict[str, Any]]: + markdown_config = markdown_config or {} + table_mode = markdown_config.get("table_mode", markdown_config.get("tableMode", "auto")) + + content = response.content.strip() + if not content: + raise ValueError("Empty message content, cannot send") + + if response_prefix: + content = f"{response_prefix} {content}" + + if table_mode == "always": + content = convert_markdown_tables(content) + elif table_mode == "auto": + content = convert_markdown_tables(content) + + chunks = chunk_text(content, limit=text_chunk_limit) + + results: list[dict[str, Any]] = [] + for idx, chunk in enumerate(chunks): + payload: dict[str, Any] = { + "token": response.identity.channel_chat_id, + "message": chunk, + "actorDisplayName": bot_display_name, + } + + if response.reply_to_message_id: + payload["replyTo"] = response.reply_to_message_id + + ref_id = response.metadata.get("reference_id") + if ref_id: + payload["referenceId"] = f"{ref_id}-{idx}" if len(chunks) > 1 else ref_id + else: + payload["referenceId"] = f"fp-{uuid.uuid4().hex}-{idx}" if len(chunks) > 1 else f"fp-{uuid.uuid4().hex}" + + if response.message_type in (MessageType.IMAGE, MessageType.FILE, MessageType.AUDIO, MessageType.VIDEO): + raise ValueError("Media messages must use send_media(), not send()") + + results.append(payload) + + return results + + +def format_outbound_single(response: ChannelResponse, bot_display_name: str) -> dict[str, Any]: + payloads = format_outbound(response, bot_display_name) + return payloads[0] if payloads else {"token": response.identity.channel_chat_id, "message": ""} diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/metrics.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/metrics.py new file mode 100644 index 00000000..535639fd --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/metrics.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import time +from collections import deque +from typing import Any + + +class NextcloudTalkMetrics: + def __init__(self, max_latency_samples: int = 100): + self._max_latency_samples = max_latency_samples + self._send_latencies: deque[float] = deque(maxlen=max_latency_samples) + self._receive_latencies: deque[float] = deque(maxlen=max_latency_samples) + self._send_count: int = 0 + self._receive_count: int = 0 + self._send_error_count: int = 0 + self._receive_error_count: int = 0 + self._circuit_breaker_trips: int = 0 + self._started_at: float = time.time() + + def record_send_success(self, latency_s: float) -> None: + self._send_latencies.append(latency_s) + self._send_count += 1 + + def record_send_error(self) -> None: + self._send_error_count += 1 + + def record_receive_success(self, latency_s: float) -> None: + self._receive_latencies.append(latency_s) + self._receive_count += 1 + + def record_receive_error(self) -> None: + self._receive_error_count += 1 + + def record_circuit_breaker_trip(self) -> None: + self._circuit_breaker_trips += 1 + + def snapshot(self) -> dict[str, Any]: + uptime_s = time.time() - self._started_at + + send_latencies = list(self._send_latencies) + recv_latencies = list(self._receive_latencies) + + avg_send = sum(send_latencies) / len(send_latencies) if send_latencies else None + avg_recv = sum(recv_latencies) / len(recv_latencies) if recv_latencies else None + + return { + "uptime_seconds": round(uptime_s, 1), + "send": { + "count": self._send_count, + "error_count": self._send_error_count, + "avg_latency_ms": round(avg_send * 1000, 1) if avg_send is not None else None, + "p50_latency_ms": round(_percentile(send_latencies, 50) * 1000, 1) if send_latencies else None, + "p95_latency_ms": round(_percentile(send_latencies, 95) * 1000, 1) if send_latencies else None, + "p99_latency_ms": round(_percentile(send_latencies, 99) * 1000, 1) if send_latencies else None, + }, + "receive": { + "count": self._receive_count, + "error_count": self._receive_error_count, + "avg_latency_ms": round(avg_recv * 1000, 1) if avg_recv is not None else None, + "p50_latency_ms": round(_percentile(recv_latencies, 50) * 1000, 1) if recv_latencies else None, + "p95_latency_ms": round(_percentile(recv_latencies, 95) * 1000, 1) if recv_latencies else None, + "p99_latency_ms": round(_percentile(recv_latencies, 99) * 1000, 1) if recv_latencies else None, + }, + "circuit_breaker": { + "trip_count": self._circuit_breaker_trips, + }, + } + + +def _percentile(data: list[float], percentile: float) -> float: + if not data: + return 0.0 + sorted_data = sorted(data) + k = (len(sorted_data) - 1) * percentile / 100.0 + f = int(k) + c = k - f + if f + 1 < len(sorted_data): + return sorted_data[f] + c * (sorted_data[f + 1] - sorted_data[f]) + return sorted_data[f] \ No newline at end of file diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/monitor_polling.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/monitor_polling.py new file mode 100644 index 00000000..36870342 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/monitor_polling.py @@ -0,0 +1,163 @@ +from __future__ import annotations + +import asyncio +import logging +from typing import Any + +import aiohttp + +from yuxi.channels.exceptions import ChannelAuthenticationError +from yuxi.channels.models import ChannelMessage + +from .client import NextcloudTalkClient +from .config import resolve_history_limit +from .normalizer import normalize_inbound + +logger = logging.getLogger(__name__) + + +async def poll_conversation( + client: NextcloudTalkClient, + api_base: str, + token: str, + last_known_msg_id: int, + channel_id: str, + timeout: int = 30, + limit: int = 20, + room_type: int | None = None, +) -> tuple[list[ChannelMessage], int]: + path = ( + f"{api_base}/chat/{token}" + f"?lookIntoFuture=1&timeout={timeout}" + f"&lastKnownMessageId={last_known_msg_id}&limit={limit}" + ) + + try: + result = await client.get(path) + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + statuscode = meta.get("statuscode") + + if statuscode == 304: + return [], last_known_msg_id + + messages = ocs.get("data", []) + if not messages or not isinstance(messages, list): + return [], last_known_msg_id + + try: + messages.sort(key=lambda m: int(m.get("id", 0))) + except (TypeError, ValueError): + pass + + channel_messages: list[ChannelMessage] = [] + new_last_id = last_known_msg_id + + for msg in messages: + msg["token"] = token + channel_msg = normalize_inbound(msg, channel_id, room_type) + channel_messages.append(channel_msg) + msg_id = int(msg.get("id", 0)) + if msg_id > new_last_id: + new_last_id = msg_id + + return channel_messages, new_last_id + + except aiohttp.ClientResponseError as e: + if e.status in (401, 403): + raise ChannelAuthenticationError() from e + elif e.status == 429: + await asyncio.sleep(5) + return [], last_known_msg_id + else: + logger.warning(f"[NextcloudTalk] Poll error for {token}: {e}") + return [], last_known_msg_id + + except Exception as e: + logger.warning(f"[NextcloudTalk] Poll error: {e}") + return [], last_known_msg_id + + +class PollingMonitor: + def __init__( + self, + client: NextcloudTalkClient, + api_base: str, + channel_id: str, + poll_interval: float = 0.5, + poll_timeout: int = 30, + config: dict[str, Any] | None = None, + ): + self._client = client + self._api_base = api_base + self._channel_id = channel_id + self._poll_interval = poll_interval + self._poll_timeout = poll_timeout + self._config = config or {} + self._last_msg_ids: dict[str, int] = {} + self._running = False + self._on_message = None + self._on_refresh = None + + def set_on_message(self, handler): + self._on_message = handler + + def set_on_refresh(self, handler): + self._on_refresh = handler + + def update_conversations(self, conversations: dict[str, Any]) -> None: + self._conversations = conversations + + async def start(self, conversations: dict[str, Any]) -> None: + self._running = True + self._conversations = conversations + for token in conversations: + self._last_msg_ids.setdefault(token, 0) + + poll_count = 0 + while self._running: + for token in list(self._conversations.keys()): + conv = self._conversations.get(token, {}) + room_type = conv.get("type") + chat_type = "direct" if room_type == 1 else "group" + history_limit = resolve_history_limit(token, self._config, chat_type) + + try: + msgs, new_last_id = await poll_conversation( + self._client, + self._api_base, + token, + self._last_msg_ids.get(token, 0), + self._channel_id, + timeout=self._poll_timeout, + limit=history_limit if history_limit > 0 else 20, + room_type=room_type, + ) + + if new_last_id > self._last_msg_ids.get(token, 0): + self._last_msg_ids[token] = new_last_id + + for msg in msgs: + if self._on_message: + await self._on_message(msg) + + except ChannelAuthenticationError: + raise + except aiohttp.ClientResponseError as e: + if e.status == 429: + await asyncio.sleep(5) + elif e.status in (401, 403): + raise ChannelAuthenticationError() from e + else: + logger.warning(f"[NextcloudTalk] Poll error for {token}: {e}") + except Exception as e: + logger.warning(f"[NextcloudTalk] Poll error: {e}") + + poll_count += 1 + if poll_count % 120 == 0 and self._on_refresh: + await self._on_refresh() + + await asyncio.sleep(self._poll_interval) + + async def stop(self) -> None: + self._running = False diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/monitor_ws.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/monitor_ws.py new file mode 100644 index 00000000..1d4308aa --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/monitor_ws.py @@ -0,0 +1,210 @@ +from __future__ import annotations + +import asyncio +import json +import logging +from typing import Any + +import aiohttp + +from .normalizer import normalize_inbound + +logger = logging.getLogger(__name__) + + +async def _fetch_signaling_settings( + server_url: str, basic_auth: str, session: aiohttp.ClientSession | None = None +) -> dict[str, Any] | None: + url = f"{server_url}/ocs/v2.php/apps/spreed/api/v3/signaling/settings" + headers = { + "Authorization": f"Basic {basic_auth}", + "OCS-APIRequest": "true", + "Accept": "application/json", + } + try: + if session is not None: + async with session.get(url, headers=headers, timeout=10) as resp: + data = await resp.json() + return data.get("ocs", {}).get("data", {}) + async with aiohttp.ClientSession() as tmp_session: + async with tmp_session.get(url, headers=headers, timeout=10) as resp: + data = await resp.json() + return data.get("ocs", {}).get("data", {}) + except Exception: + return None + + +async def try_connect_websocket( + server_url: str, + basic_auth: str, + bot_user: str, + app_password: str, + session: aiohttp.ClientSession | None = None, +): + try: + import websockets + except ImportError: + logger.info("[NextcloudTalk] websockets package not installed, skipping WebSocket") + return None + + ws_url = server_url.replace("https://", "wss://").replace("http://", "ws://") + ws_url = f"{ws_url}/spreed/signaling" + + hello_msg: dict[str, Any] + + signaling = await _fetch_signaling_settings(server_url, basic_auth, session) + if signaling and signaling.get("helloAuthParams"): + hello_params = signaling["helloAuthParams"] + version_key = next(iter(hello_params), "1.0") + hello_msg = { + "type": "hello", + "hello": { + "version": version_key, + "auth": hello_params[version_key], + }, + } + else: + hello_msg = { + "type": "hello", + "hello": { + "version": "1.0", + "auth": { + "url": server_url, + "params": f"user={bot_user}&password={app_password}", + }, + }, + } + + try: + ws = await websockets.connect( + ws_url, + extra_headers={"Authorization": f"Basic {basic_auth}"}, + open_timeout=10, + ) + + await ws.send(json.dumps(hello_msg)) + resp_raw = await asyncio.wait_for(ws.recv(), timeout=5) + resp = json.loads(resp_raw) + + if not isinstance(resp, dict): + logger.warning("[NextcloudTalk] Hello response is not a JSON object") + await ws.close() + return None + + resp_type = resp.get("type") + if resp_type != "hello": + logger.warning(f"[NextcloudTalk] Unexpected hello response type: {resp_type}, expected 'hello'") + await ws.close() + return None + + hello_data = resp.get("hello", {}) + if not isinstance(hello_data, dict): + logger.warning("[NextcloudTalk] Hello response 'hello' field is not an object") + await ws.close() + return None + + server_version = hello_data.get("version", "unknown") + logger.info(f"[NextcloudTalk] WebSocket connected, server version: {server_version}") + return ws + + except (TimeoutError, Exception) as e: + logger.info(f"[NextcloudTalk] WebSocket not available ({e}), will use polling") + return None + + +class WebSocketMonitor: + def __init__(self, ws, channel_id: str, basic_auth: str, server_url: str, bot_user: str, app_password: str): + self._ws = ws + self._channel_id = channel_id + self._basic_auth = basic_auth + self._server_url = server_url + self._bot_user = bot_user + self._app_password = app_password + self._running = False + self._on_message = None + self._on_fallback = None + + def set_on_message(self, handler): + self._on_message = handler + + def set_on_fallback(self, handler): + self._on_fallback = handler + + async def run(self) -> None: + self._running = True + reconnect_count = 0 + reconnect_delay = 1.0 + + while self._running: + try: + raw = await asyncio.wait_for(self._ws.recv(), timeout=60) + data = json.loads(raw) + + if data.get("type") == "event" and data.get("event", {}).get("target") == "room": + await self._handle_event(data.get("event", {})) + elif data.get("type") == "ping": + await self._ws.send(json.dumps({"type": "pong"})) + else: + logger.debug(f"[NextcloudTalk] WS message: {data.get('type')}") + + reconnect_count = 0 + reconnect_delay = 1.0 + + except (TimeoutError, Exception) as e: + logger.warning(f"[NextcloudTalk] WebSocket disconnected: {e}") + reconnect_count += 1 + if reconnect_count > 3: + logger.warning("[NextcloudTalk] WebSocket failed 3 times, falling back to polling") + self._running = False + if self._on_fallback: + await self._on_fallback() + return + + await asyncio.sleep(reconnect_delay) + reconnect_delay = min(reconnect_delay * 2, 30) + + try: + self._ws = await try_connect_websocket( + self._server_url, + self._basic_auth, + self._bot_user, + self._app_password, + ) + if self._ws is None: + if self._on_fallback: + await self._on_fallback() + return + except Exception: + if self._on_fallback: + await self._on_fallback() + return + + async def _handle_event(self, event: dict[str, Any]) -> None: + event_type = event.get("type", "") + + if event_type == "chat": + chat_data = event.get("chat", {}) + if isinstance(chat_data, dict): + for key in ("refresh", "message"): + if key in chat_data: + msg_data = chat_data[key] + if isinstance(msg_data, dict): + channel_msg = normalize_inbound(msg_data, self._channel_id) + if self._on_message: + await self._on_message(channel_msg) + + elif event_type == "join": + pass + elif event_type == "leave": + pass + else: + logger.debug(f"[NextcloudTalk] WS event: {event_type}") + + async def stop(self) -> None: + self._running = False + if self._ws: + try: + await self._ws.close() + except Exception: + pass + self._ws = None diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/normalize.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/normalize.py new file mode 100644 index 00000000..0f39b3ad --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/normalize.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import logging +import re + +logger = logging.getLogger(__name__) + +CHANNEL_PREFIX = "nextcloud-talk" + +_PURE_TOKEN_PATTERN = re.compile(r"^[a-z0-9]{20,64}$") + + +def is_pure_token(target: str) -> bool: + return bool(_PURE_TOKEN_PATTERN.match(target.strip())) + + +def resolve_target_with_hint(target: str) -> tuple[str, dict[str, str]]: + target = target.strip().lower() + hints: dict[str, str] = {} + if is_pure_token(target): + hints["resolver_hint"] = "direct_token" + return f"{CHANNEL_PREFIX}:{target}", hints + if target.startswith("room:"): + hints["resolver_hint"] = "room_token" + elif target.startswith("user:"): + hints["resolver_hint"] = "user_id" + result = normalize_nc_target(target) + return result, hints + + +def normalize_nc_target(target: str) -> str: + target = target.strip().lower() + for prefix in ("nextcloud-talk:", "nextcloudtalk:", "nc-talk:", "nc:"): + if target.startswith(prefix): + target = target[len(prefix) :] + break + return f"{CHANNEL_PREFIX}:{target}" + + +def strip_nc_prefix(target: str) -> str: + target = target.strip().lower() + for prefix in ("nextcloud-talk:", "nextcloudtalk:", "nc-talk:", "nc:"): + if target.startswith(prefix): + return target[len(prefix) :] + return target + + +def looks_like_nc_target(target_id: str) -> bool: + target_id = target_id.strip().lower() + prefixes = ("nextcloud-talk:", "nextcloudtalk:", "nc-talk:", "nc:") + return any(target_id.startswith(p) for p in prefixes) + + +def normalize_room_target(token: str) -> str: + return f"{CHANNEL_PREFIX}:room:{token}" + + +def normalize_user_target(user_id: str) -> str: + return f"{CHANNEL_PREFIX}:{user_id}" diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/normalizer.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/normalizer.py new file mode 100644 index 00000000..9e9b8da6 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/normalizer.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +from datetime import datetime, UTC +from typing import Any + +from yuxi.channels.models import ( + Attachment, + ChannelIdentity, + ChannelMessage, + ChannelType, + ChatType, + EventType, + MentionsInfo, + MessageType, +) + + +_AS2_TYPE_MAP: dict[str, EventType] = { + "Create": EventType.MESSAGE_RECEIVED, + "Announce": EventType.MESSAGE_RECEIVED, + "Update": EventType.MESSAGE_UPDATED, + "Delete": EventType.MESSAGE_DELETED, + "Like": EventType.SYSTEM_EVENT, + "Dislike": EventType.SYSTEM_EVENT, + "Add": EventType.MEMBER_JOINED, + "Remove": EventType.MEMBER_LEFT, + "Join": EventType.MEMBER_JOINED, + "Leave": EventType.MEMBER_LEFT, +} + + +def _parse_as2_activity(raw: dict[str, Any]) -> dict[str, Any] | None: + if "@context" not in raw: + return None + + as2_type = raw.get("type", "") + obj = raw.get("object", {}) + if isinstance(obj, dict): + content = obj.get("content", "") + obj_id = obj.get("id", "") + message_type = obj.get("messageType", raw.get("messageType", "comment")) + else: + content = raw.get("message", "") + obj_id = "" + message_type = raw.get("messageType", "comment") + + actor = raw.get("actor", {}) + if isinstance(actor, dict): + actor_id = actor.get("id", "") + actor_display_name = actor.get("name", "") + else: + actor_id = raw.get("actorId", "") + actor_display_name = raw.get("actorDisplayName", "") + + target = raw.get("target", {}) + if isinstance(target, dict): + token = target.get("id", "") + else: + token = raw.get("token", "") + + return { + "id": obj_id or raw.get("id", ""), + "token": token or raw.get("token", ""), + "actorId": actor_id or raw.get("actorId", ""), + "actorDisplayName": actor_display_name or raw.get("actorDisplayName", ""), + "timestamp": raw.get("published", raw.get("timestamp", 0)), + "message": content or raw.get("message", ""), + "messageType": message_type, + "_as2_type": as2_type, + } + + +def normalize_inbound(raw: dict[str, Any], channel_id: str, room_type: int | None = None) -> ChannelMessage: + as2_parsed = _parse_as2_activity(raw) + as2_type = "" + if as2_parsed: + raw = {**raw, **{k: v for k, v in as2_parsed.items() if not k.startswith("_")}} + as2_type = as2_parsed["_as2_type"] + + msg_type = raw.get("messageType", raw.get("message_type", "comment")) + + identity = ChannelIdentity( + channel_id=channel_id, + channel_type=ChannelType.NEXTCLOUDTALK, + channel_user_id=raw.get("actorId", raw.get("actor_id", "")), + channel_chat_id=raw.get("token", ""), + channel_message_id=str(raw.get("id", "")), + ) + + if msg_type == "system": + return _build_system_event(raw, identity, channel_id, as2_parsed) + + message_type = MessageType.COMMAND if msg_type == "command" else MessageType.TEXT + content = raw.get("message", "") + + attachments = [] + params = raw.get("messageParameters", {}) + for param_data in params.values(): + ptype = param_data.get("type", "") + if ptype in ("file", "media"): + attachments.append( + Attachment( + type="file", + url=param_data.get("link", ""), + filename=param_data.get("name", ""), + mime_type=param_data.get("mimetype", ""), + size_bytes=param_data.get("size", 0), + ) + ) + elif ptype == "talk-poll": + message_type = MessageType.POLL + + reply_to = None + parent_data = raw.get("parent") + if parent_data: + reply_to = str(parent_data.get("id", "")) + + timestamp = datetime.fromtimestamp(raw.get("timestamp", 0), tz=UTC) + + metadata = { + "nc_message_type": msg_type, + "nc_actor_type": raw.get("actorType", raw.get("actor_type", "")), + "nc_actor_display_name": raw.get("actorDisplayName", raw.get("actor_display_name", "")), + "reference_id": raw.get("referenceId", raw.get("reference_id", "")), + "nc_message_parameters": params, + } + if as2_parsed: + metadata["as2_type"] = as2_parsed["_as2_type"] + + was_mentioned = False + sender_name = raw.get("actorDisplayName", raw.get("actor_display_name", "")) + for key, param_data in params.items(): + if key.startswith("mention"): + param_type = param_data.get("type", "") + if param_type in ("user", "guest"): + was_mentioned = True + elif param_data.get("type") == "user" and param_data.get("id") == identity.channel_user_id: + pass + + is_forwarded = msg_type == "comment" and "{mention-forwarded}" in content + if is_forwarded: + forward_actor = raw.get("forwardActorDisplayName", raw.get("forward_actor_display_name", "")) + forward_actor_id = raw.get("forwardActorId", raw.get("forward_actor_id", "")) + metadata["nc_forwarded"] = True + metadata["nc_forward_actor_display_name"] = forward_actor + metadata["nc_forward_actor_id"] = forward_actor_id + + mentions_info = MentionsInfo( + mentioned_user_ids=[], + is_bot_mentioned=was_mentioned, + raw_text=content, + ) + + chat_type = ChatType.DIRECT + if room_type is not None and room_type in (2, 3): + chat_type = ChatType.GROUP + + return ChannelMessage( + identity=identity, + event_type=EventType.MESSAGE_RECEIVED, + message_type=message_type, + chat_type=chat_type, + content=content, + attachments=attachments, + mentions=mentions_info, + reply_to_message_id=reply_to, + metadata=metadata, + timestamp=timestamp, + ) + + +def _build_system_event( + raw: dict[str, Any], identity: ChannelIdentity, channel_id: str, as2_parsed: dict[str, Any] | None = None +) -> ChannelMessage: + system_msg = raw.get("systemMessage", raw.get("system_message", "")) + + event_map = { + "user_added": EventType.MEMBER_JOINED, + "user_removed": EventType.MEMBER_LEFT, + "conversation_created": EventType.SYSTEM_EVENT, + "conversation_renamed": EventType.SYSTEM_EVENT, + "conversation_description_changed": EventType.SYSTEM_EVENT, + "call_started": EventType.SYSTEM_EVENT, + "call_ended": EventType.SYSTEM_EVENT, + "call_missed": EventType.SYSTEM_EVENT, + "message_deleted": EventType.MESSAGE_DELETED, + "reaction": EventType.SYSTEM_EVENT, + "reaction_revoked": EventType.SYSTEM_EVENT, + "reaction_added": EventType.SYSTEM_EVENT, + "reaction_removed": EventType.SYSTEM_EVENT, + "poll_created": EventType.SYSTEM_EVENT, + "poll_closed": EventType.SYSTEM_EVENT, + "poll_voted": EventType.SYSTEM_EVENT, + "conversation_password_changed": EventType.SYSTEM_EVENT, + "lobby_timer_reached": EventType.SYSTEM_EVENT, + "lobby_none": EventType.SYSTEM_EVENT, + "breakout_rooms_started": EventType.SYSTEM_EVENT, + "breakout_rooms_stopped": EventType.SYSTEM_EVENT, + "recording_started": EventType.SYSTEM_EVENT, + "recording_stopped": EventType.SYSTEM_EVENT, + "avatar_updated": EventType.SYSTEM_EVENT, + "message_edited": EventType.MESSAGE_UPDATED, + "message_pinned": EventType.SYSTEM_EVENT, + "message_unpinned": EventType.SYSTEM_EVENT, + } + + event_type = event_map.get(system_msg, EventType.SYSTEM_EVENT) + description = system_msg if system_msg else "system" + + metadata: dict[str, Any] = {"nc_system_message": system_msg} + if as2_parsed: + metadata["as2_type"] = as2_parsed["_as2_type"] + + return ChannelMessage( + identity=identity, + event_type=event_type, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=description, + metadata=metadata, + ) diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/pairing.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/pairing.py new file mode 100644 index 00000000..76245e9a --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/pairing.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import logging +from typing import Any + +from .client import NextcloudTalkClient + +logger = logging.getLogger(__name__) + +PAIRING_MESSAGE = ( + "Hello! Your account is not yet authorized to interact with this bot.\n\n" + "Your Nextcloud user id: `{user_id}`\n\n" + "Please ask an administrator to approve your access by adding your user id " + "to the paired users list using:\n\n" + "```\n" + "from yuxi.channels.adapters.nextcloudtalk.security import pair_user\n" + 'pair_user("{user_id}")\n' + "```\n\n" + "Once approved, send another message to start interacting with the bot." +) + +APPROVED_MESSAGE = "Your account `{user_id}` has been approved! Send a message to start." + + +async def send_pairing_challenge( + client: NextcloudTalkClient, + api_base: str, + token: str, + user_id: str, + config: dict[str, Any] | None = None, +) -> bool: + try: + message = PAIRING_MESSAGE.format(user_id=user_id) + payload = { + "token": token, + "message": message, + "referenceId": f"pairing-{user_id}", + } + result = await client.post(f"{api_base}/chat/{token}", json_data=payload) + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + if meta.get("statuscode") == 201: + logger.info(f"[NextcloudTalk] Pairing challenge sent to {user_id} in {token}") + return True + logger.warning(f"[NextcloudTalk] Failed to send pairing challenge: {meta}") + return False + except Exception as e: + logger.warning(f"[NextcloudTalk] Failed to send pairing challenge: {e}") + return False + + +async def send_pairing_approved( + client: NextcloudTalkClient, + api_base: str, + token: str, + user_id: str, + config: dict[str, Any] | None = None, +) -> bool: + try: + message = APPROVED_MESSAGE.format(user_id=user_id) + payload = { + "token": token, + "message": message, + "referenceId": f"pairing-approved-{user_id}", + } + result = await client.post(f"{api_base}/chat/{token}", json_data=payload) + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + if meta.get("statuscode") == 201: + logger.info(f"[NextcloudTalk] Pairing approved notification sent to {user_id}") + return True + return False + except Exception as e: + logger.warning(f"[NextcloudTalk] Failed to send pairing approved: {e}") + return False diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/probe.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/probe.py new file mode 100644 index 00000000..f94c419f --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/probe.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import logging +import time +from typing import Any + +from aiohttp import ClientSession + +logger = logging.getLogger(__name__) + +CAPABILITIES_ENDPOINT = "/ocs/v2.php/cloud/capabilities" + +_CAPABILITIES_CACHE: dict[str, tuple[float, dict[str, Any]]] = {} +_CACHE_TTL = 300 + + +async def probe_capabilities(http: ClientSession, server_url: str) -> dict[str, Any]: + now = time.time() + cached = _CAPABILITIES_CACHE.get(server_url) + if cached and now - cached[0] < _CACHE_TTL: + return cached[1] + + url = f"{server_url}{CAPABILITIES_ENDPOINT}" + headers = {"OCS-APIRequest": "true", "Accept": "application/json"} + + try: + async with http.get(url, headers=headers, timeout=10) as resp: + data = await resp.json() + ocs = data.get("ocs", {}) + caps = ocs.get("data", {}).get("capabilities", {}) + spreed = caps.get("spreed", {}) + if not spreed: + logger.warning("[NextcloudTalk] spreed capability not found in response") + _CAPABILITIES_CACHE[server_url] = (now, spreed) + return spreed + except Exception as e: + logger.warning(f"[NextcloudTalk] capabilities probe failed: {e}") + return cached[1] if cached else {} + + +def invalidate_cache(server_url: str | None = None) -> None: + if server_url: + _CAPABILITIES_CACHE.pop(server_url, None) + else: + _CAPABILITIES_CACHE.clear() + + +def resolve_api_version(capabilities: dict[str, Any]) -> str: + version = capabilities.get("version", "") + if not version: + return "v1" + + try: + major = int(version.split(".")[0]) + except (ValueError, IndexError): + return "v1" + + if major >= 17: + return "v4" + elif major >= 12: + return "v3" + return "v1" + + +def resolve_api_base(api_version: str) -> str: + return f"/ocs/v2.php/apps/spreed/api/{api_version}" diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/room_info.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/room_info.py new file mode 100644 index 00000000..78fe4c21 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/room_info.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import logging +import time +from typing import Any + +from yuxi.channels.models import ChatType + +logger = logging.getLogger(__name__) + +_CACHE_TTL_SUCCESS = 300 +_CACHE_TTL_ERROR = 30 + +_room_kind_cache: dict[str, tuple[float, str | None]] = {} + + +async def resolve_room_kind( + client, + api_base: str, + token: str, + api_user: str = "", + api_password: str = "", +) -> dict[str, Any]: + now = time.time() + cached = _room_kind_cache.get(token) + if cached: + cached_ts, cached_kind = cached + if cached_kind is not None and now - cached_ts < _CACHE_TTL_SUCCESS: + return _kind_result(cached_kind) + if cached_kind is None and now - cached_ts < _CACHE_TTL_ERROR: + return _kind_result(None) + + try: + path = f"{api_base}/room/{token}" + result = await client.get(path) + ocs = result.get("ocs", {}) + data = ocs.get("data", {}) + room_type = data.get("type") + kind = _map_room_type(room_type) + _room_kind_cache[token] = (now, kind) + return _kind_result(kind) + except Exception as e: + logger.warning(f"[NextcloudTalk] Room type query failed for {token}: {e}") + _room_kind_cache[token] = (now, None) + return _kind_result(None) + + +def _map_room_type(room_type: int | None) -> str | None: + if room_type is None: + return None + if room_type in (1, 5, 6): + return "direct" + if room_type in (2, 3, 4): + return "group" + return None + + +def _kind_result(kind: str | None) -> dict[str, Any]: + if kind == "direct": + return {"kind": "direct", "chat_type": ChatType.DIRECT} + if kind == "group": + return {"kind": "group", "chat_type": ChatType.GROUP} + return {"kind": None, "chat_type": ChatType.GROUP, "source": "cache_fallback"} + + +def invalidate_room_cache(token: str | None = None) -> None: + if token: + _room_kind_cache.pop(token, None) + else: + _room_kind_cache.clear() diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/secret_contract.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/secret_contract.py new file mode 100644 index 00000000..4c250509 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/secret_contract.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import logging +import os +from typing import Any + + +logger = logging.getLogger(__name__) + +SECRET_TARGETS = [ + {"key": "bot_user", "label": "Bot Username", "source": "direct", "secret": False}, + {"key": "app_password", "label": "App Password", "source": "direct", "secret": True}, + {"key": "api_user", "label": "API Username", "source": "direct", "secret": False}, + {"key": "api_password", "label": "API Password", "source": "direct", "secret": True}, + {"key": "bot_secret_file", "label": "Bot Secret File", "source": "file", "secret": True}, + {"key": "NEXTCLOUD_TALK_BOT_USER", "label": "Env: NEXTCLOUD_TALK_BOT_USER", "source": "env", "secret": False}, + { + "key": "NEXTCLOUD_TALK_APP_PASSWORD", + "label": "Env: NEXTCLOUD_TALK_APP_PASSWORD", + "source": "env", + "secret": True, + }, + { + "key": "NEXTCLOUD_TALK_API_PASSWORD", + "label": "Env: NEXTCLOUD_TALK_API_PASSWORD", + "source": "env", + "secret": True, + }, + { + "key": "NEXTCLOUD_TALK_BOT_SECRET", + "label": "Env: NEXTCLOUD_TALK_BOT_SECRET", + "source": "env", + "secret": True, + }, +] + + +def list_secret_targets() -> list[dict[str, Any]]: + return SECRET_TARGETS + + +def collect_runtime_secrets(config: dict[str, Any]) -> dict[str, Any]: + secrets: dict[str, Any] = {} + bot_user = config.get("bot_user", config.get("botUser", "")) + app_password = config.get("app_password", config.get("appPassword", "")) + api_user = config.get("api_user", "") + api_password = config.get("api_password", "") + + if not bot_user: + bot_user = os.getenv("NEXTCLOUD_TALK_BOT_USER", "") + if not app_password: + app_password = os.getenv( + "NEXTCLOUD_TALK_APP_PASSWORD", + os.getenv("NEXTCLOUD_TALK_API_PASSWORD", os.getenv("NEXTCLOUD_TALK_BOT_SECRET", "")), + ) + + secrets["bot_user"] = {"value": bot_user, "present": bool(bot_user)} + secrets["app_password"] = {"value": "***" if app_password else "", "present": bool(app_password)} + secrets["api_user"] = {"value": api_user, "present": bool(api_user)} + secrets["api_password"] = {"value": "***" if api_password else "", "present": bool(api_password)} + + return secrets + + +def verify_secret_contract(config: dict[str, Any]) -> dict[str, Any]: + result: dict[str, Any] = {"valid": True, "issues": []} + secrets = collect_runtime_secrets(config) + + if not secrets["bot_user"]["present"]: + result["valid"] = False + result["issues"].append("bot_user is missing") + if not secrets["app_password"]["present"]: + result["valid"] = False + result["issues"].append("app_password is missing") + + server_url = config.get("server_url", "") + if not server_url: + result["valid"] = False + result["issues"].append("server_url is missing") + + return result diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/security.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/security.py new file mode 100644 index 00000000..3c6e6a34 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/security.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +import json +import logging +import os +import re +from typing import Any +from urllib.parse import urlparse + +logger = logging.getLogger(__name__) + +_PERSIST_PATH = os.getenv( + "NEXTCLOUDTALK_PAIRED_USERS_FILE", + os.path.join(os.path.dirname(__file__), ".paired_users.json"), +) +_PAIRED_USERS: set[str] | None = None + + +def _get_paired_users() -> set[str]: + global _PAIRED_USERS + if _PAIRED_USERS is None: + _PAIRED_USERS = set() + _load_paired_users() + return _PAIRED_USERS + + +def _load_paired_users() -> None: + try: + if os.path.exists(_PERSIST_PATH): + with open(_PERSIST_PATH, encoding="utf-8") as f: + data = json.load(f) + if isinstance(data, list): + _get_paired_users().update(data) + except Exception: + pass + + +def _save_paired_users() -> None: + try: + with open(_PERSIST_PATH, "w", encoding="utf-8") as f: + json.dump(sorted(_get_paired_users()), f) + except Exception as e: + logger.warning(f"[NextcloudTalk] Failed to persist paired users: {e}") + + +def check_dm_policy(user_id: str, config: dict[str, Any]) -> bool: + dm_policy = config.get("dm_policy", "pairing") + + match dm_policy: + case "open": + return True + case "disabled": + return False + case "pairing": + return _is_user_paired(user_id) + case "allowlist": + allow_from = config.get("allow_from", config.get("allowFrom", [])) + return f"nc:{user_id}" in allow_from + case _: + return False + + +def check_group_policy(token: str, user_id: str, config: dict[str, Any]) -> bool: + group_policy = config.get("group_policy", "allowlist") + + match group_policy: + case "open": + return True + case "disabled": + return False + case "allowlist": + groups = config.get("groups", {}) + conv_config = groups.get(token, {}) + per_group_allow = conv_config.get("allow_from", conv_config.get("allowFrom", [])) + if f"nc:{user_id}" in per_group_allow: + return True + global_allow = config.get("group_allow_from", config.get("groupAllowFrom", [])) + return f"nc:{user_id}" in global_allow + case _: + return False + + +def _is_user_paired(user_id: str) -> bool: + return user_id in _get_paired_users() + + +def pair_user(user_id: str) -> None: + _get_paired_users().add(user_id) + _save_paired_users() + + +def unpair_user(user_id: str) -> None: + _get_paired_users().discard(user_id) + _save_paired_users() + + +def check_mention_gate( + token: str, + content: str, + bot_name: str, + config: dict[str, Any], + raw_message_params: dict[str, Any] | None = None, +) -> bool: + require_mention = True + + rooms_cfg = config.get("rooms", {}) + room_cfg = rooms_cfg.get(token, rooms_cfg.get("*", {})) + if "requireMention" in room_cfg: + require_mention = room_cfg["requireMention"] + elif "require_mention" in room_cfg: + require_mention = room_cfg["require_mention"] + + if not require_mention: + return True + + mention_check_content = content.lower() + mention_check_bot_name = bot_name.lower() + + if raw_message_params: + for key, param in raw_message_params.items(): + if key.startswith("mention"): + ptype = param.get("type", "") + param_name = param.get("name", "") + if ptype in ("user", "guest"): + if param_name and param_name.lower() == mention_check_bot_name: + return True + + if f"@{mention_check_bot_name}" in mention_check_content: + return True + + logger.debug(f"[NextcloudTalk] Mention gate blocked: bot '{bot_name}' not mentioned in group {token}") + return False + + +def validate_nextcloud_url(url: str, server_url: str, audit_context: str = "") -> bool: + _ctx = f"[{audit_context}] " if audit_context else "" + + if re.search(r"/\.{1,2}/", url): + logger.warning(f"[NextcloudTalk] {_ctx}SSRF blocked: path traversal detected in {url}") + return False + + parsed = urlparse(url) + server_parsed = urlparse(server_url) + + if parsed.scheme not in ("https", "http"): + logger.warning(f"[NextcloudTalk] {_ctx}SSRF blocked: invalid scheme in {url}") + return False + + if parsed.hostname != server_parsed.hostname: + logger.warning( + f"[NextcloudTalk] {_ctx}SSRF blocked: external host {parsed.hostname} != {server_parsed.hostname}" + ) + return False + + if parsed.port and parsed.port != server_parsed.port and parsed.port not in (80, 443): + logger.warning(f"[NextcloudTalk] {_ctx}SSRF blocked: port mismatch {parsed.port}") + return False + + return True + + +def validate_backend_source( + headers: dict[str, str], + server_url: str, + audit_context: str = "", +) -> bool: + _ctx = f"[{audit_context}] " if audit_context else "" + backend_header = headers.get("X-Nextcloud-Talk-Backend", headers.get("x-nextcloud-talk-backend", "")) + if not backend_header: + return False + + parsed_backend = urlparse(backend_header) + parsed_server = urlparse(server_url) + + if parsed_backend.hostname != parsed_server.hostname: + logger.warning( + f"[NextcloudTalk] {_ctx}Backend source mismatch: {parsed_backend.hostname} != {parsed_server.hostname}" + ) + return False + + return True + + +def collect_security_warnings(config: dict[str, Any]) -> list[str]: + warnings: list[str] = [] + + group_policy = config.get("group_policy", "allowlist") + if group_policy == "open": + warnings.append("group_policy is set to 'open' - bot responds to all group messages") + + dm_policy = config.get("dm_policy", "pairing") + if dm_policy == "open": + warnings.append("dm_policy is set to 'open' - bot responds to all DM messages without pairing") + + groups = config.get("groups", {}) + if not groups and group_policy == "allowlist": + warnings.append( + "No per-room allowlist configured - group messages will be blocked unless global allowlist is set" + ) + + allow_from = config.get("allow_from", config.get("allowFrom", [])) + if dm_policy == "allowlist" and not allow_from: + warnings.append("DM allowlist policy active but allow_from is empty - all DMs will be blocked") + + rooms_cfg = config.get("rooms", {}) + if not rooms_cfg: + warnings.append("No per-room configuration - mention gate defaults to requireMention=true for all group chats") + + return warnings diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/send.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/send.py new file mode 100644 index 00000000..5a6b5ef8 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/send.py @@ -0,0 +1,295 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import random +from collections.abc import Callable +from typing import Any + +import aiohttp + +from yuxi.channels.exceptions import ( + ChannelAuthenticationError, + ChannelRateLimitError, + DeliveryFailedError, +) +from yuxi.channels.models import DeliveryResult + +from .client import NextcloudTalkClient + +logger = logging.getLogger(__name__) + + +def _retry_config(config: dict[str, Any]) -> dict[str, Any]: + retry_cfg = config.get("retry", {}) + return { + "max_attempts": retry_cfg.get("attempts", 3), + "min_delay_ms": retry_cfg.get("min_delay_ms", 400), + "max_delay_ms": retry_cfg.get("max_delay_ms", 30000), + "jitter": retry_cfg.get("jitter", 0.1), + } + + +async def _with_retry( + func: Callable[[], Any], + config: dict[str, Any], +) -> Any: + rc = _retry_config(config) + for attempt in range(rc["max_attempts"]): + try: + return await func() + except aiohttp.ClientResponseError as e: + if e.status == 401: + raise ChannelAuthenticationError() from e + if e.status == 429: + retry_after = int(e.headers.get("Retry-After", "5")) + if attempt < rc["max_attempts"] - 1: + await asyncio.sleep(retry_after) + else: + raise ChannelRateLimitError(retry_after_ms=retry_after * 1000) from e + elif e.status in (404, 403): + return DeliveryResult(success=False, error=f"HTTP {e.status}: {e.message}") + elif attempt >= rc["max_attempts"] - 1: + raise DeliveryFailedError(str(e)) from e + except Exception as e: + if attempt >= rc["max_attempts"] - 1: + raise DeliveryFailedError(str(e)) from e + + delay = min(rc["min_delay_ms"] * (2**attempt), rc["max_delay_ms"]) / 1000 + delay += delay * rc["jitter"] * (random.random() * 2 - 1) + await asyncio.sleep(delay) + + return DeliveryResult(success=False, error="Max retries exceeded") + + +async def send_text( + client: NextcloudTalkClient, + api_base: str, + token: str, + payload: dict[str, Any], + config: dict[str, Any], + sent_cache: dict[str, str] | None = None, + silent: bool = False, +) -> DeliveryResult: + reference_id = payload.get("referenceId", "") + if sent_cache and reference_id and reference_id in sent_cache: + logger.debug(f"[NextcloudTalk] Duplicate send prevented for ref {reference_id}") + return DeliveryResult(success=True, message_id=sent_cache[reference_id]) + + async def _do(): + json_data = dict(payload) + if silent: + json_data["silent"] = True + result = await client.post(f"{api_base}/chat/{token}", json_data=json_data) + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + if meta.get("statuscode") == 201: + msg_data = ocs.get("data", {}) + msg_id = str(msg_data.get("id", "")) + if sent_cache is not None and reference_id and msg_id: + sent_cache[reference_id] = msg_id + return DeliveryResult(success=True, message_id=msg_id) + return DeliveryResult(success=False, error=meta.get("message", "Unknown error")) + + return await _with_retry(_do, config) + + +async def edit_message( + client: NextcloudTalkClient, + api_base: str, + token: str, + message_id: str, + new_content: str, + config: dict[str, Any] | None = None, +) -> DeliveryResult: + config = config or {} + + async def _do(): + result = await client.put( + f"{api_base}/chat/{token}/{message_id}", + json_data={"message": new_content}, + ) + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + if meta.get("statuscode") in (200, 201): + return DeliveryResult(success=True, message_id=message_id) + return DeliveryResult(success=False, error=meta.get("message", "Unknown error")) + + return await _with_retry(_do, config) + + +async def delete_message( + client: NextcloudTalkClient, + api_base: str, + token: str, + message_id: str, + config: dict[str, Any] | None = None, +) -> DeliveryResult: + config = config or {} + + async def _do(): + result = await client.delete(f"{api_base}/chat/{token}/{message_id}") + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + if meta.get("statuscode") in (200, 201): + return DeliveryResult(success=True, message_id=message_id) + return DeliveryResult(success=False, error=meta.get("message", "Unknown error")) + + return await _with_retry(_do, config) + + +async def send_media( + client: NextcloudTalkClient, + api_base: str, + chat_id: str, + media_type: str, + media_data: bytes, + config: dict[str, Any] | None = None, + **kwargs, +) -> DeliveryResult: + config = config or {} + filename = kwargs.get("filename", f"attachment.{media_type.split('/')[-1]}") + caption = kwargs.get("caption", "") + + async def _do(): + form = aiohttp.FormData() + form.add_field("file", media_data, filename=filename, content_type=media_type) + form.add_field("metaData", json.dumps({"message": caption})) + result = await client.post_form(f"{api_base}/chat/{chat_id}/share", form) + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + if meta.get("statuscode") == 201: + msg_data = ocs.get("data", {}) + return DeliveryResult(success=True, message_id=str(msg_data.get("id", ""))) + return DeliveryResult(success=False, error=meta.get("message", "Unknown error")) + + return await _with_retry(_do, config) + + +async def send_media_chunked( + client: NextcloudTalkClient, + api_base: str, + chat_id: str, + media_type: str, + media_data: bytes, + chunk_size: int = 5 * 1024 * 1024, + config: dict[str, Any] | None = None, + on_progress: Callable[[int, int], None] | None = None, + **kwargs, +) -> DeliveryResult: + total_size = len(media_data) + if total_size <= chunk_size: + return await send_media(client, api_base, chat_id, media_type, media_data, config, **kwargs) + + config = config or {} + filename = kwargs.get("filename", f"attachment.{media_type.split('/')[-1]}") + caption = kwargs.get("caption", "") + uploaded = 0 + + async def _do(): + nonlocal uploaded + result = await client.put( + f"{api_base}/chat/{chat_id}/upload/init", + json_data={"filename": filename, "size": total_size, "content_type": media_type}, + ) + ocs = result.get("ocs", {}) + if ocs.get("meta", {}).get("statuscode") != 201: + return DeliveryResult(success=False, error="Failed to initialize chunked upload") + upload_id = ocs.get("data", {}).get("id", "") + + while uploaded < total_size: + end = min(uploaded + chunk_size, total_size) + chunk = media_data[uploaded:end] + form = aiohttp.FormData() + form.add_field("chunk", chunk, filename=filename, content_type=media_type) + form.add_field("offset", str(uploaded)) + result = await client.post_form(f"{api_base}/chat/{chat_id}/upload/{upload_id}/chunk", form) + ocs = result.get("ocs", {}) + if ocs.get("meta", {}).get("statuscode") != 201: + return DeliveryResult(success=False, error=f"Chunk upload failed at offset {uploaded}") + uploaded = end + if on_progress: + on_progress(uploaded, total_size) + + meta_payload = {"metaData": json.dumps({"message": caption})} + result = await client.post( + f"{api_base}/chat/{chat_id}/upload/{upload_id}/finalize", + json_data=meta_payload, + ) + ocs = result.get("ocs", {}) + if ocs.get("meta", {}).get("statuscode") == 201: + msg_data = ocs.get("data", {}) + return DeliveryResult(success=True, message_id=str(msg_data.get("id", ""))) + return DeliveryResult(success=False, error=ocs.get("meta", {}).get("message", "Finalize failed")) + + return await _with_retry(_do, config) + + +async def send_reaction( + client: NextcloudTalkClient, + api_base: str, + token: str, + message_id: str, + emoji: str, + config: dict[str, Any] | None = None, +) -> DeliveryResult: + config = config or {} + + async def _do(): + result = await client.post( + f"{api_base}/reaction/{token}/{message_id}", + json_data={"reaction": emoji}, + ) + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + if meta.get("statuscode") in (200, 201): + return DeliveryResult(success=True, message_id=message_id) + return DeliveryResult(success=False, error=meta.get("message", "Unknown error")) + + return await _with_retry(_do, config) + + +async def send_reaction_delete( + client: NextcloudTalkClient, + api_base: str, + token: str, + message_id: str, + emoji: str, + config: dict[str, Any] | None = None, +) -> DeliveryResult: + config = config or {} + + async def _do(): + result = await client.delete( + f"{api_base}/reaction/{token}/{message_id}", + json_data={"reaction": emoji}, + ) + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + if meta.get("statuscode") in (200, 201): + return DeliveryResult(success=True, message_id=message_id) + return DeliveryResult(success=False, error=meta.get("message", "Unknown error")) + + return await _with_retry(_do, config) + + +async def get_reactions( + client: NextcloudTalkClient, + api_base: str, + token: str, + message_id: str, + config: dict[str, Any] | None = None, +) -> DeliveryResult: + config = config or {} + + async def _do(): + result = await client.get(f"{api_base}/reaction/{token}/{message_id}") + ocs = result.get("ocs", {}) + meta = ocs.get("meta", {}) + if meta.get("statuscode") in (200, 201): + data = ocs.get("data", {}) + return DeliveryResult(success=True, message_id=message_id, metadata={"reactions": data}) + return DeliveryResult(success=False, error=meta.get("message", "Unknown error")) + + return await _with_retry(_do, config) diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/session.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/session.py new file mode 100644 index 00000000..6afdabc4 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/session.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import logging +from typing import Any + +from yuxi.channels.models import ChatType + +logger = logging.getLogger(__name__) + + +def resolve_session_route( + token: str, + conversation_info: dict[str, Any] | None, + default_agent_id: str, +) -> dict[str, str]: + room_type = conversation_info.get("type") if conversation_info else None + + if room_type == 1: + route = f"agent:{default_agent_id}:nextcloudtalk:dm:{token}" + else: + route = f"agent:{default_agent_id}:nextcloudtalk:group:{token}" + + return {"route": route, "token": token} + + +def resolve_chat_type(room_type: int | None) -> ChatType: + if room_type == 1: + return ChatType.DIRECT + if room_type in (2, 3): + return ChatType.GROUP + return ChatType.GROUP + + +def make_thread_key(agent_id: str, token: str, room_type: int | None) -> str: + if room_type == 1: + return f"agent:{agent_id}:nextcloudtalk:dm:{token}" + return f"agent:{agent_id}:nextcloudtalk:group:{token}" + + +def resolve_conversation_label(conv_info: dict[str, Any]) -> str: + room_type = conv_info.get("type") if conv_info else None + display_name = conv_info.get("display_name", conv_info.get("displayName", "")) if conv_info else "" + name = conv_info.get("name", "") if conv_info else "" + + if room_type == 1: + return display_name or name or "Direct Message" + label = display_name or name or "Group" + return f"room:{label}" diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/setup_wizard.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/setup_wizard.py new file mode 100644 index 00000000..300bdafc --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/setup_wizard.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +WIZARD_STEPS = [ + { + "step": 0, + "title": "Bot Installation Guide", + "description": ( + "Before configuring the adapter, install a Talk bot on your Nextcloud server.\n\n" + "Run this command as an administrator on your Nextcloud server:\n\n" + "```bash\n" + "occ talk:bot:install \n" + "```\n\n" + "Example:\n```bash\n" + 'occ talk:bot:install "ForcePilot Bot" my_secret_key https://your-server.example.com\n' + "```\n\n" + "After installation, use the bot username and the secret as app_password in Step 2." + ), + "field": None, + "required": False, + }, + { + "step": 1, + "title": "Nextcloud Instance URL", + "description": "Enter the base URL of your Nextcloud server (e.g., https://cloud.example.com).", + "field": "server_url", + "required": True, + }, + { + "step": 2, + "title": "Bot Credentials", + "description": ( + "Enter the bot username and application password. " + "Create an app password in Nextcloud at Settings > Security.\n\n" + "Environment variables (alternative to config):\n" + "- `NEXTCLOUD_TALK_BOT_USER` for bot_user\n" + "- `NEXTCLOUD_TALK_APP_PASSWORD` or `NEXTCLOUD_TALK_API_PASSWORD` for app_password" + ), + "fields": ["bot_user", "app_password"], + "required": True, + "preferredEnvVar": { + "app_password": "NEXTCLOUD_TALK_API_PASSWORD", + }, + }, + { + "step": 3, + "title": "API Credentials (Optional)", + "description": "Optionally provide separate API credentials for room/peer queries. Leave blank to reuse bot credentials.", + "fields": ["api_user", "api_password"], + "required": False, + }, + { + "step": 4, + "title": "DM Policy", + "description": "Choose how the bot handles direct messages: open (anyone), pairing (require admin approval), allowlist (specified users only), or disabled.", + "field": "dm_policy", + "required": True, + "options": [ + {"value": "open", "label": "Open - anyone can message the bot"}, + {"value": "pairing", "label": "Pairing - users must be approved by admin"}, + {"value": "allowlist", "label": "Allowlist - only specified users"}, + {"value": "disabled", "label": "Disabled - no DMs allowed"}, + ], + }, + { + "step": 5, + "title": "Allow From", + "description": "Configure allowed user IDs and room tokens. Format: nc:{user_id} for users.", + "field": "allow_from", + "required": False, + }, +] + + +def generate_wizard_steps() -> list[dict[str, Any]]: + return WIZARD_STEPS + + +def validate_step(step_num: int, data: dict[str, Any]) -> dict[str, Any]: + step = next((s for s in WIZARD_STEPS if s["step"] == step_num), None) + if not step: + return {"valid": False, "error": f"Unknown step: {step_num}"} + + errors: list[str] = [] + + if step_num == 1: + server_url = data.get("server_url", "").strip() + if step["required"] and not server_url: + errors.append("Server URL is required") + elif server_url and not (server_url.startswith("https://") or server_url.startswith("http://")): + errors.append("Server URL must start with https:// or http://") + + elif step_num == 2: + bot_user = data.get("bot_user", "").strip() + app_password = data.get("app_password", "").strip() + if not bot_user: + errors.append("Bot username is required") + if not app_password: + errors.append("App password is required") + + elif step_num == 3: + pass + + elif step_num == 4: + dm_policy = data.get("dm_policy", "").strip() + valid_policies = {"open", "pairing", "allowlist", "disabled"} + if dm_policy and dm_policy not in valid_policies: + errors.append(f"DM policy must be one of: {', '.join(sorted(valid_policies))}") + + elif step_num == 5: + allow_from = data.get("allow_from", []) + if isinstance(allow_from, list): + for entry in allow_from: + if not isinstance(entry, str) or not entry.startswith("nc:"): + errors.append(f"Each allow_from entry must start with 'nc:', got: {entry}") + + if errors: + return {"valid": False, "errors": errors} + return {"valid": True} + + +def build_config_from_wizard(wizard_data: dict[str, Any]) -> dict[str, Any]: + config: dict[str, Any] = { + "server_url": wizard_data.get("server_url", "").rstrip("/"), + "bot_user": wizard_data.get("bot_user", ""), + "app_password": wizard_data.get("app_password", ""), + "dm_policy": wizard_data.get("dm_policy", "pairing"), + "try_websocket": True, + } + + api_user = wizard_data.get("api_user", "") + if api_user: + config["api_user"] = api_user + api_password = wizard_data.get("api_password", "") + if api_password: + config["api_password"] = api_password + + allow_from = wizard_data.get("allow_from", []) + if allow_from: + config["allow_from"] = allow_from + + return config diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/status.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/status.py new file mode 100644 index 00000000..f752b512 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/status.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import logging + +from yuxi.channels.models import ChannelAccountSnapshot, TokenStatus + +logger = logging.getLogger(__name__) + + +def build_snapshot(adapter) -> ChannelAccountSnapshot: + config = adapter.config or {} + server_url = config.get("server_url", "") + + bot_user = config.get("bot_user", config.get("botUser", "")) + has_bot_creds = bool(bot_user and (config.get("app_password") or config.get("appPassword"))) + + credential_source = "none" + if config.get("app_password") or config.get("appPassword"): + credential_source = "config" + elif config.get("bot_secret_file"): + credential_source = "secretFile" + + bot_token_source = "config" if has_bot_creds else "none" + api_user = config.get("api_user", "") + api_password = config.get("api_password", "") + has_api_creds = bool(api_user and api_password) + app_token_source = "config" if has_api_creds else "none" + + dm_policy = config.get("dm_policy", "pairing") + group_policy = config.get("group_policy", "allowlist") + allow_from = config.get("allow_from", config.get("allowFrom", [])) + group_allow_from = config.get("group_allow_from", config.get("groupAllowFrom", [])) + + monitor_mode = getattr(adapter, "_monitor_mode", "polling") + status_value = getattr(adapter, "_status", None) + status_str = str(status_value.value) if hasattr(status_value, "value") else str(status_value or "disconnected") + + conversations = getattr(adapter, "_conversations", {}) + + return ChannelAccountSnapshot( + account_id=getattr(adapter, "channel_id", "nextcloud-talk"), + name="Nextcloud Talk", + configured=has_bot_creds, + enabled=True, + linked=has_bot_creds, + running=status_str == "connected", + connected=status_str == "connected", + status_state="configured" if has_bot_creds else "not-configured", + health_state=status_str, + dm_policy=dm_policy, + group_policy=group_policy, + allow_from_count=len(allow_from) + len(group_allow_from), + base_url=server_url, + secret_source=credential_source, + credential_source=credential_source, + bot_token_source=bot_token_source, + bot_token_status=TokenStatus(source=bot_token_source, status="ok" if has_bot_creds else "missing"), + app_token_source=app_token_source, + app_token_status=TokenStatus(source=app_token_source, status="ok" if has_api_creds else "missing"), + probe={ + "monitor_mode": monitor_mode, + "conversation_count": len(conversations), + "talk_version": getattr(adapter, "_talk_version", ""), + "api_base": getattr(adapter, "_api_base", ""), + }, + ) diff --git a/backend/package/yuxi/channels/adapters/nextcloudtalk/webhook_server.py b/backend/package/yuxi/channels/adapters/nextcloudtalk/webhook_server.py new file mode 100644 index 00000000..fe5c19d6 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/nextcloudtalk/webhook_server.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import json +import logging +import time +from collections import defaultdict +from http import HTTPStatus +from typing import Any + +from aiohttp import web + +from .client import verify_hmac_signature +from .normalizer import normalize_inbound +from .security import validate_backend_source + +logger = logging.getLogger(__name__) + +DEFAULT_WEBHOOK_PORT = 8788 +DEFAULT_WEBHOOK_HOST = "0.0.0.0" +DEFAULT_WEBHOOK_PATH = "/nextcloud-talk-webhook" +DEFAULT_RATE_WINDOW_S = 60 +DEFAULT_RATE_MAX_REQUESTS = 10 +DEFAULT_RATE_LOCKOUT_S = 300 +DEFAULT_MAX_BODY_SIZE = 1 * 1024 * 1024 +DEFAULT_PRE_AUTH_MAX_BODY_SIZE = 64 * 1024 + + +class _AuthRateLimiter: + def __init__( + self, + max_requests: int = DEFAULT_RATE_MAX_REQUESTS, + window_s: int = DEFAULT_RATE_WINDOW_S, + lockout_s: int = DEFAULT_RATE_LOCKOUT_S, + ): + self._max_requests = max_requests + self._window_s = window_s + self._lockout_s = lockout_s + self._failures: dict[str, list[float]] = defaultdict(list) + self._lockouts: dict[str, float] = {} + + def _gc(self, now: float) -> None: + stale = [ip for ip, lockout_until in self._lockouts.items() if now >= lockout_until] + for ip in stale: + del self._lockouts[ip] + cutoff = now - self._window_s + for ip in list(self._failures): + self._failures[ip] = [ts for ts in self._failures[ip] if ts > cutoff] + if not self._failures[ip]: + del self._failures[ip] + + def check(self, client_ip: str) -> bool: + now = time.time() + self._gc(now) + + if client_ip in self._lockouts: + if now < self._lockouts[client_ip]: + return False + del self._lockouts[client_ip] + + if len(self._failures.get(client_ip, [])) >= self._max_requests: + self._lockouts[client_ip] = now + self._lockout_s + logger.warning(f"[NextcloudTalk] Auth rate limiter locked out {client_ip} for {self._lockout_s}s") + return False + + return True + + def record_failure(self, client_ip: str) -> None: + self._failures[client_ip].append(time.time()) + + def stats(self) -> dict[str, Any]: + self._gc(time.time()) + return { + "tracked_ips": len(self._failures), + "locked_ips": len(self._lockouts), + } + + +class NextcloudTalkWebhookServer: + def __init__( + self, + adapter, + port: int = DEFAULT_WEBHOOK_PORT, + host: str = DEFAULT_WEBHOOK_HOST, + path: str = DEFAULT_WEBHOOK_PATH, + rate_limit_max_requests: int = DEFAULT_RATE_MAX_REQUESTS, + rate_limit_window_s: int = DEFAULT_RATE_WINDOW_S, + rate_limit_lockout_s: int = DEFAULT_RATE_LOCKOUT_S, + max_body_size: int = DEFAULT_MAX_BODY_SIZE, + pre_auth_max_body_size: int = DEFAULT_PRE_AUTH_MAX_BODY_SIZE, + ): + self._adapter = adapter + self._port = port + self._host = host + self._path = path.rstrip("/") if path != "/" else path + self._max_body_size = max_body_size + self._pre_auth_max_body_size = pre_auth_max_body_size + self._app: web.Application | None = None + self._runner: web.AppRunner | None = None + self._site: web.TCPSite | None = None + self._rate_limiter = _AuthRateLimiter( + max_requests=rate_limit_max_requests, + window_s=rate_limit_window_s, + lockout_s=rate_limit_lockout_s, + ) + + @property + def port(self) -> int: + return self._port + + @property + def host(self) -> str: + return self._host + + @property + def path(self) -> str: + return self._path + + @property + def public_url(self) -> str: + host = self._host if self._host != "0.0.0.0" else "localhost" + port_part = f":{self._port}" if self._port not in (80, 443) else "" + return f"http://{host}{port_part}{self._path}" + + def rate_limiter_stats(self) -> dict[str, Any]: + return self._rate_limiter.stats() + + async def start(self) -> None: + self._app = web.Application() + self._app.router.add_get("/healthz", self._handle_healthz) + self._app.router.add_post(self._path, self._handle_webhook) + + self._runner = web.AppRunner(self._app) + await self._runner.setup() + self._site = web.TCPSite(self._runner, self._host, self._port) + await self._site.start() + logger.info(f"[NextcloudTalk] Webhook server listening on {self._host}:{self._port}{self._path}") + + async def stop(self) -> None: + if self._runner: + await self._runner.cleanup() + self._runner = None + self._site = None + self._app = None + logger.info("[NextcloudTalk] Webhook server stopped") + + async def _handle_healthz(self, request: web.Request) -> web.Response: + return web.json_response( + { + "status": "ok", + "channel": self._adapter.channel_id, + "rate_limiter": self._rate_limiter.stats(), + } + ) + + async def _handle_webhook(self, request: web.Request) -> web.Response: + client_ip = request.remote or "unknown" + + content_length = request.content_length + if content_length is not None and content_length > self._max_body_size: + logger.warning(f"[NextcloudTalk] Body too large from {client_ip}: {content_length} > {self._max_body_size}") + return web.Response(status=HTTPStatus.PAYLOAD_TOO_LARGE, text="Request body too large") + + try: + body = await request.read() + if len(body) > self._max_body_size: + logger.warning(f"[NextcloudTalk] Body oversized from {client_ip}: {len(body)} > {self._max_body_size}") + return web.Response(status=HTTPStatus.PAYLOAD_TOO_LARGE, text="Request body too large") + except Exception: + return web.Response(status=HTTPStatus.BAD_REQUEST, text="Cannot read body") + + if len(body) > self._pre_auth_max_body_size: + logger.warning( + f"[NextcloudTalk] Body exceeds pre-auth limit from {client_ip}: " + f"{len(body)} > {self._pre_auth_max_body_size}" + ) + return web.Response(status=HTTPStatus.PAYLOAD_TOO_LARGE, text="Request body too large") + + headers = {k.lower(): v for k, v in request.headers.items()} + + if not self._verify_request(headers, body): + self._rate_limiter.record_failure(client_ip) + return web.Response(status=HTTPStatus.UNAUTHORIZED, text="Signature verification failed") + + backend_header = headers.get("x-nextcloud-talk-backend", "") + server_url = self._adapter.config.get("server_url", "") + if ( + backend_header + and server_url + and not validate_backend_source( + {"X-Nextcloud-Talk-Backend": backend_header}, server_url, audit_context="nextcloud-talk-webhook" + ) + ): + logger.warning(f"[NextcloudTalk] Backend source validation failed for {client_ip}") + return web.Response(status=HTTPStatus.FORBIDDEN, text="Backend source mismatch") + + try: + payload = json.loads(body) + except json.JSONDecodeError: + return web.Response(status=HTTPStatus.BAD_REQUEST, text="Invalid JSON") + + try: + message = normalize_inbound(payload, self._adapter.channel_id) + except Exception as e: + logger.warning(f"[NextcloudTalk] Failed to normalize webhook payload: {e}") + return web.Response(status=HTTPStatus.BAD_REQUEST, text="Invalid payload format") + + try: + await self._adapter._handle_message(message) + except Exception as e: + logger.error(f"[NextcloudTalk] Failed to process webhook message: {e}", exc_info=True) + return web.Response(status=HTTPStatus.INTERNAL_SERVER_ERROR, text="Message processing failed") + + return web.Response(status=HTTPStatus.OK, text="OK") + + def _verify_request(self, headers: dict[str, str], body: bytes) -> bool: + client_ip = headers.get("x-forwarded-for", headers.get("x-real-ip", "unknown")) + if not self._rate_limiter.check(client_ip): + return False + + signature = headers.get("x-nextcloud-talk-sha256", "") + if not signature: + return False + + bot_secret = self._adapter.config.get("app_password", self._adapter.config.get("appPassword", "")) + if not bot_secret: + logger.warning("[NextcloudTalk] Cannot verify webhook: no bot_secret configured") + return False + + return verify_hmac_signature(body, signature, bot_secret)