ForcePilot/backend/package/yuxi/channel/extensions/nextcloud_talk/send.py
Kris 6fc9a4a94c feat(channel): 添加 Nextcloud Talk 渠道扩展
新增 Nextcloud Talk 渠道扩展,支持在 Yuxi 平台中集成 Nextcloud Talk 即时通讯渠道。

包含以下功能模块:
- accounts: 账户管理
- bot_admin: Bot 管理
- chat_api: 聊天 API 封装
- config_schema: 配置模式
- conversation: 会话管理
- gateway: SSE/WebSocket 网关接入
- inbound: 入站消息处理
- send: 消息发送
- webhook_server: Webhook 服务
- format: 消息格式转换
- normalize: 消息规范化
- poll_api: 投票 API
- reaction_api: 表情反应 API
- policy: 安全策略
- probe: 健康探测
- replay_guard: 重放防护
- signature: 签名验证
- room_info: 房间信息
- types: 类型定义
2026-05-21 11:29:56 +08:00

156 lines
4.7 KiB
Python

from __future__ import annotations
import ipaddress
import logging
import secrets
from urllib.parse import urljoin, urlparse
import httpx
from yuxi.channel.extensions.nextcloud_talk.format import convert_markdown_tables_to_ascii
from yuxi.channel.extensions.nextcloud_talk.normalize import strip_nextcloud_talk_target_prefix
from yuxi.channel.extensions.nextcloud_talk.signature import generate_nextcloud_talk_signature
logger = logging.getLogger(__name__)
OCS_MESSAGE_PATH = "/ocs/v2.php/apps/spreed/api/v1/bot/{token}/message"
OCS_REACTION_PATH = "/ocs/v2.php/apps/spreed/api/v1/bot/{token}/reaction/{message_id}"
class NextcloudTalkSendError(Exception):
def __init__(self, message: str, status_code: int | None = None):
super().__init__(message)
self.status_code = status_code
def _check_ssrf(base_url: str, dangerously_allow_private_network: bool = False) -> None:
if dangerously_allow_private_network:
return
try:
parsed = urlparse(base_url)
except ValueError:
return
hostname = parsed.hostname
if not hostname:
return
from yuxi.channel.security.ssrf_guard import SsrfGuard
guard = SsrfGuard(default_deny=False)
try:
ip_str = str(ipaddress.ip_address(hostname))
except ValueError:
return
ip_result = guard.check_ip(ip_str)
if not ip_result.safe:
raise NextcloudTalkSendError(
f"SSRF blocked: {hostname}{ip_result.reason}. Set dangerously_allow_private_network=true to allow."
)
async def send_message_nextcloud_talk(
to: str,
text: str,
base_url: str,
secret: str,
*,
reply_to: str | None = None,
silent: bool = False,
reference_id: str | None = None,
timeout: float = 30.0,
dangerously_allow_private_network: bool = False,
) -> dict:
_check_ssrf(base_url, dangerously_allow_private_network)
room_token = strip_nextcloud_talk_target_prefix(to)
if not text.strip():
raise NextcloudTalkSendError("Message must be non-empty")
text = convert_markdown_tables_to_ascii(text)
body = {"message": text}
if reply_to:
body["replyTo"] = reply_to
if silent:
body["silent"] = True
if reference_id is None:
reference_id = secrets.token_hex(16)
body["referenceId"] = reference_id
random, signature = generate_nextcloud_talk_signature(text, secret)
url = urljoin(base_url.rstrip("/") + "/", OCS_MESSAGE_PATH.format(token=room_token))
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(
url,
json=body,
headers={
"Content-Type": "application/json",
"OCS-APIRequest": "true",
"X-Nextcloud-Talk-Bot-Random": random,
"X-Nextcloud-Talk-Bot-Signature": signature,
"Accept": "application/json",
},
)
if resp.status_code == 200:
data = resp.json()
ocs_data = data.get("ocs", {}).get("data", {})
return {
"message_id": str(ocs_data.get("id", "unknown")),
"room_token": room_token,
"timestamp": ocs_data.get("timestamp"),
"reference_id": reference_id,
}
error_body = resp.text[:500]
status_map = {
400: f"Nextcloud Talk: bad request - {error_body}",
401: "Nextcloud Talk: authentication failed - check bot secret",
403: "Nextcloud Talk: forbidden - bot may not have permission in this room",
404: f"Nextcloud Talk: room not found (token={room_token})",
}
msg = status_map.get(resp.status_code, f"Nextcloud Talk send failed: {error_body}")
raise NextcloudTalkSendError(msg, resp.status_code)
async def send_reaction_nextcloud_talk(
room_token: str,
message_id: str,
reaction: str,
base_url: str,
secret: str,
*,
timeout: float = 30.0,
dangerously_allow_private_network: bool = False,
) -> bool:
_check_ssrf(base_url, dangerously_allow_private_network)
random, signature = generate_nextcloud_talk_signature(reaction, secret)
url = urljoin(
base_url.rstrip("/") + "/",
OCS_REACTION_PATH.format(token=room_token, message_id=message_id),
)
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(
url,
json={"reaction": reaction},
headers={
"Content-Type": "application/json",
"OCS-APIRequest": "true",
"X-Nextcloud-Talk-Bot-Random": random,
"X-Nextcloud-Talk-Bot-Signature": signature,
"Accept": "application/json",
},
)
return resp.status_code == 200