feat(backend): add Nextcloud Talk channel adapter implementation
实现了完整的Nextcloud Talk聊天适配器，包含以下核心功能： 1. 基础客户端通信与认证 2. 会话路由与聊天类型解析 3. 消息分块与格式转换 4. 用户配对与权限校验 5. 轮询式消息监听 6. 配置验证与诊断工具 7. 安全策略与去重缓存 8. 指标统计与状态快照 新增配套的配对用户缓存文件与模块导出入口。
This commit is contained in:
parent
bd60c15df0
commit
b018ad25da
@ -0,0 +1 @@
|
||||
[]
|
||||
@ -0,0 +1,3 @@
|
||||
from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter
|
||||
|
||||
__all__ = ["NextcloudTalkAdapter"]
|
||||
@ -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
|
||||
1018
backend/package/yuxi/channels/adapters/nextcloudtalk/adapter.py
Normal file
1018
backend/package/yuxi/channels/adapters/nextcloudtalk/adapter.py
Normal file
File diff suppressed because it is too large
Load Diff
@ -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}"}
|
||||
@ -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
|
||||
121
backend/package/yuxi/channels/adapters/nextcloudtalk/client.py
Normal file
121
backend/package/yuxi/channels/adapters/nextcloudtalk/client.py
Normal file
@ -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()
|
||||
@ -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))
|
||||
@ -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)
|
||||
@ -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"
|
||||
127
backend/package/yuxi/channels/adapters/nextcloudtalk/dedup.py
Normal file
127
backend/package/yuxi/channels/adapters/nextcloudtalk/dedup.py
Normal file
@ -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()
|
||||
@ -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()
|
||||
]
|
||||
187
backend/package/yuxi/channels/adapters/nextcloudtalk/doctor.py
Normal file
187
backend/package/yuxi/channels/adapters/nextcloudtalk/doctor.py
Normal file
@ -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)}
|
||||
@ -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": ""}
|
||||
@ -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]
|
||||
@ -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
|
||||
@ -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
|
||||
@ -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}"
|
||||
@ -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,
|
||||
)
|
||||
@ -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
|
||||
@ -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}"
|
||||
@ -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()
|
||||
@ -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
|
||||
209
backend/package/yuxi/channels/adapters/nextcloudtalk/security.py
Normal file
209
backend/package/yuxi/channels/adapters/nextcloudtalk/security.py
Normal file
@ -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
|
||||
295
backend/package/yuxi/channels/adapters/nextcloudtalk/send.py
Normal file
295
backend/package/yuxi/channels/adapters/nextcloudtalk/send.py
Normal file
@ -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)
|
||||
@ -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}"
|
||||
@ -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 <bot_name> <bot_secret> <bot_url>\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
|
||||
@ -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", ""),
|
||||
},
|
||||
)
|
||||
@ -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)
|
||||
Loading…
Reference in New Issue
Block a user