diff --git a/docs/ENVIRONMENT_VARIABLES.md b/docs/ENVIRONMENT_VARIABLES.md index fda84c74..240a8188 100644 --- a/docs/ENVIRONMENT_VARIABLES.md +++ b/docs/ENVIRONMENT_VARIABLES.md @@ -181,6 +181,21 @@ - 作用:`tool_log` 中 args/result 截断上限 - 生效:`oclaw/platform/persistence/sqlite_store.py` +- `AIA_MAX_ATTACHMENT_BYTES` + - 默认:`26214400`(25MB) + - 作用:工具结果/MCP payload 中嵌入式 base64 内容落盘为 `attachment_id` 时的单附件最大字节数(超限则不落盘,降级为 `*_ref` 元信息并标记 `attachment_too_large`) + - 取值:`0` 表示不限制(不推荐) + - 生效:`oclaw/runtime/chat/media_redact.py` + +- `AIA_ATTACHMENT_ACL_STRICT` + - 默认:`0` + - 作用:附件下载鉴权是否严格依赖 `attachment_acl` + - 说明: + - `0`:优先走 `attachment_acl`,缺失时仍允许回退扫描历史 `chat_message.attachments`(兼容旧数据) + - `1`:**严格模式**,只允许 `attachment_acl`(以及用户头像 `avatar_attachment_id`)命中的附件被下载 + - 上线建议:先执行 Admin “ACL 回填”,再开启 strict + - 生效:`oclaw/interfaces/admin/chat_api.py` + - `AIA_IMAGE_TOOL_RESULT_REPLAY_CAP_CHARS` - 默认:`4000` - 作用:限制历史轮次中 `query_image_attachment`(OCR/描述)结果回放到模型上下文时的 `text` 长度上限 diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index 9d2effc5..a1508762 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -356,6 +356,26 @@ npm install images-mcp powershell -ExecutionPolicy Bypass -File .\scripts\weixin_install.ps1 ``` +注意: + +- 脚本支持两种模式: + - 官方插件模式:`-UseOpenclawCli`(不校验 `runner.ts/login.ts`) + - 本地 sidecar 模式:`-LocalSourcePath`(会强校验 `runner.ts/login.ts`) + +- 默认按本地 sidecar 模式处理:必须传 `-LocalSourcePath`(指向你们自研 weixin sidecar 源码目录)。 + +```powershell +powershell -ExecutionPolicy Bypass -File .\scripts\weixin_install.ps1 -LocalSourcePath "D:\path\to\your-weixin-module" +``` + +- 若你要临时切回官方链路,可显式指定: + +```powershell +powershell -ExecutionPolicy Bypass -File .\scripts\weixin_install.ps1 -UseOpenclawCli +``` + +- 仅本地 sidecar 模式会强校验 `runner.ts` / `login.ts` 是否存在;官方插件模式不做该校验。 + 安装目录: - `data/channel_sidecar/oclaw-weixin/` diff --git a/docs/attachment-acl.md b/docs/attachment-acl.md new file mode 100644 index 00000000..392f5ca1 --- /dev/null +++ b/docs/attachment-acl.md @@ -0,0 +1,72 @@ +# Attachment ACL(附件访问控制) + +本文档说明 admin chat 附件下载鉴权的访问控制策略、回填流程与 strict 模式上线建议。 + +## 背景 + +系统把工具结果/上传文件等二进制内容落盘为 `attachment_id`(内容 hash),前端通过: + +- `GET /admin/api/chat/attachments/{attachment_id}` + +获取附件 bytes(下载/预览)。 + +为避免仅凭 `attachment_id` 造成越权读取,需要对下载接口做归属校验。 + +## 数据结构 + +- **`attachment_acl`** + - 用途:记录附件归属关系(哪个租户/用户/会话以何种来源产生/引用了该附件) + - 主键:`(attachment_id, tenant_id, user_id, session_id, source)` + +## 鉴权策略 + +下载接口会校验: + +- `attachment_id` 格式合法(64 位 hex) +- 当前登录用户是否被 ACL 授权访问该附件 +- 头像附件通过 `avatar_attachment_id` 放行(不走 ACL) + +## strict 模式 + +环境变量: + +- `AIA_ATTACHMENT_ACL_STRICT=1` + +行为: + +- 下载鉴权只信 `attachment_acl`(以及头像 `avatar_attachment_id`) +- **不会**回退扫描历史 `chat_message.attachments` + +适用场景: + +- 生产环境附件访问收口 +- 已完成历史数据 ACL 回填 + +## 回填流程(Admin) + +为了让历史消息中的附件也能被 strict 模式识别,需要先回填 ACL。 + +入口(仅 `administrator` 可见/可调用): + +- 管理台 Chat 页面右上角菜单:**回填 ACL** +- 或接口:`POST /admin/api/chat/admin/attachments/acl/backfill?limit_messages=...` + +建议流程: + +1. 在低峰期执行回填(默认扫描最近 50k 条含 attachments 的消息) +2. 观察返回结果(扫描/插入计数) +3. 开启 `AIA_ATTACHMENT_ACL_STRICT=1` + +## 相关配置放一起(ENV) + +- `AIA_MAX_ATTACHMENT_BYTES`:控制工具结果 base64 落盘单附件大小上限 +- `AIA_ATTACHMENT_ACL_STRICT`:下载鉴权严格只信 `attachment_acl` + +## 回滚建议 + +如 strict 模式误伤历史附件下载(403): + +1. 临时关闭 strict:`AIA_ATTACHMENT_ACL_STRICT=0` +2. 再次执行回填(提高 `limit_messages`) +3. 重新开启 strict + diff --git a/interfaces/admin/chat_api.py b/interfaces/admin/chat_api.py index ec7ebfdb..4edc2bd1 100644 --- a/interfaces/admin/chat_api.py +++ b/interfaces/admin/chat_api.py @@ -7,6 +7,8 @@ import json import queue import threading import os +import re +import time from collections.abc import Callable, Iterator from typing import Any from pathlib import Path @@ -66,6 +68,11 @@ _CHAT_MSG_LIMIT = 256 _SESSION_TITLE_MAX_LEN = 120 _AVATAR_UPLOAD_MAX_BYTES = 2 * 1024 * 1024 _AVATAR_MIMES = frozenset({"image/png", "image/jpeg", "image/jpg", "image/webp", "image/gif"}) +_ATTACHMENT_ID_RE = re.compile(r"^[a-f0-9]{64}$") +_ATT_DOWNLOAD_BUCKET: dict[str, tuple[float, float]] = {} +_ATT_DOWNLOAD_LOCK = threading.Lock() +_ATT_DOWNLOAD_RATE_PER_SEC = 1.0 # 60/min +_ATT_DOWNLOAD_BURST = 20.0 _CHAT_STOP_EVENTS: dict[str, threading.Event] = {} _CHAT_STOP_LOCK = threading.Lock() @@ -252,6 +259,58 @@ def _parse_attachments_payload(raw: Any) -> list[dict[str, Any]] | None: return out if out else None +def _is_valid_attachment_id(raw: str) -> bool: + return bool(_ATTACHMENT_ID_RE.fullmatch(str(raw or "").strip().lower())) + + +def _can_access_attachment(store: SqliteStore, ctx: dict[str, Any], *, attachment_id: str) -> bool: + aid = str(attachment_id or "").strip().lower() + if not _is_valid_attachment_id(aid): + return False + tenant_id = str(ctx.get("tenant_id") or "").strip() + user_id = str(ctx.get("user_id") or "").strip() + if not tenant_id: + return False + if _is_administrator_chat_viewer(ctx): + if store.attachment_acl_allows_tenant(tenant_id=tenant_id, attachment_id=aid): + return True + if (not _attachment_acl_strict()) and store.attachment_referenced_in_tenant(tenant_id=tenant_id, attachment_id=aid): + return True + # Keep profile/avatar behavior compatible for administrator account. + admin = store.get_user_by_username(tenant_id=tenant_id, username="administrator") or {} + return str(admin.get("avatar_attachment_id") or "").strip().lower() == aid + if user_id and store.attachment_acl_allows_user(tenant_id=tenant_id, user_id=user_id, attachment_id=aid): + return True + if (not _attachment_acl_strict()) and user_id and store.attachment_referenced_by_user(tenant_id=tenant_id, user_id=user_id, attachment_id=aid): + return True + user = store.get_user_by_id(tenant_id=tenant_id, user_id=user_id) if user_id else None + return str((user or {}).get("avatar_attachment_id") or "").strip().lower() == aid + + +def _rate_limit_attachment_download(*, actor_tenant_id: str, actor_user_id: str) -> bool: + tid = str(actor_tenant_id or "").strip() + uid = str(actor_user_id or "").strip() + if not tid or not uid: + return False + key = f"{tid}:{uid}" + now = time.time() + with _ATT_DOWNLOAD_LOCK: + tokens, last = _ATT_DOWNLOAD_BUCKET.get(key, (_ATT_DOWNLOAD_BURST, now)) + dt = max(0.0, now - float(last or now)) + tokens = min(_ATT_DOWNLOAD_BURST, float(tokens) + dt * _ATT_DOWNLOAD_RATE_PER_SEC) + if tokens < 1.0: + _ATT_DOWNLOAD_BUCKET[key] = (tokens, now) + return False + tokens -= 1.0 + _ATT_DOWNLOAD_BUCKET[key] = (tokens, now) + return True + + +def _attachment_acl_strict() -> bool: + raw = str(os.getenv("AIA_ATTACHMENT_ACL_STRICT") or "").strip().lower() + return raw in {"1", "true", "yes", "on"} + + def _chat_username(ctx: dict[str, Any]) -> str: return str(ctx.get("username") or "").strip().lower() @@ -1448,18 +1507,86 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor authorization: str | None = Header(default=None), ) -> Response: store = SqliteStore(db_path()) - _ = resolve_auth(store, authorization) - aid = str(attachment_id or "").strip() + ctx = resolve_auth(store, authorization) + tenant_id = str(ctx.get("tenant_id") or "").strip() + user_id = str(ctx.get("user_id") or "").strip() + aid = str(attachment_id or "").strip().lower() if not aid: raise HTTPException(status_code=400, detail="attachment_id_required") + if not _is_valid_attachment_id(aid): + raise HTTPException(status_code=400, detail="attachment_id_invalid") + if not _rate_limit_attachment_download(actor_tenant_id=tenant_id, actor_user_id=user_id): + store.add_admin_audit_log( + actor_tenant_id=tenant_id, + actor_user_id=user_id, + action="chat_attachment_download", + target_type="attachment", + target_id=aid, + status="rate_limited", + detail={"path": "chat_api.attachment_bytes"}, + ) + raise HTTPException(status_code=429, detail="rate_limited") + if not _can_access_attachment(store, ctx, attachment_id=aid): + store.add_admin_audit_log( + actor_tenant_id=tenant_id, + actor_user_id=user_id, + action="chat_attachment_download", + target_type="attachment", + target_id=aid, + status="forbidden", + detail={"path": "chat_api.attachment_bytes"}, + ) + raise HTTPException(status_code=403, detail="attachment_forbidden") ast = AttachmentAssetStore() try: blob, meta = ast.load_bytes(aid) except Exception: + store.add_admin_audit_log( + actor_tenant_id=tenant_id, + actor_user_id=user_id, + action="chat_attachment_download", + target_type="attachment", + target_id=aid, + status="not_found", + detail={"path": "chat_api.attachment_bytes"}, + ) raise HTTPException(status_code=404, detail="attachment_not_found") from None mime = (meta.mime if meta else None) or "application/octet-stream" + store.add_admin_audit_log( + actor_tenant_id=tenant_id, + actor_user_id=user_id, + action="chat_attachment_download", + target_type="attachment", + target_id=aid, + status="ok", + detail={"path": "chat_api.attachment_bytes", "mime": mime, "bytes": int(meta.bytes if meta else len(blob))}, + ) return Response(content=blob, media_type=mime) + @chat.post("/admin/attachments/acl/backfill") + def api_chat_admin_backfill_attachment_acl( + limit_messages: int = Query(default=50_000, ge=1, le=500_000), + authorization: str | None = Header(default=None), + ) -> dict[str, Any]: + store = SqliteStore(db_path()) + ctx = resolve_auth(store, authorization) + _require_administrator_chat_viewer(ctx) + tenant_id = str(ctx.get("tenant_id") or "").strip() + user_id = str(ctx.get("user_id") or "").strip() + res = store.backfill_attachment_acl_from_messages(tenant_id=tenant_id, limit_messages=int(limit_messages)) + if bool(res.get("ok")): + res.setdefault("recommended_next", "Set AIA_ATTACHMENT_ACL_STRICT=1 after verifying downloads.") + store.add_admin_audit_log( + actor_tenant_id=tenant_id, + actor_user_id=user_id, + action="attachment_acl_backfill", + target_type="tenant", + target_id=tenant_id, + status="ok" if bool(res.get("ok")) else "error", + detail=res, + ) + return res + @chat.post("/sessions/{session_id}/messages") def api_chat_send( session_id: str, @@ -1507,6 +1634,20 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor text = _effective_user_text(text=text_raw, attachments=attachments, store=store) if not _is_administrator_chat_viewer(ctx) and tenant_id and user_id: store.ensure_ui_session_owner(session_id=session_id, tenant_id=tenant_id, user_id=user_id) + # Record attachment ownership for access control (download endpoint). + if attachments and tenant_id and user_id: + for a in attachments: + if not isinstance(a, dict): + continue + aid = str(a.get("attachment_id") or "").strip().lower() + if aid and _is_valid_attachment_id(aid): + store.link_attachment_acl( + tenant_id=tenant_id, + user_id=user_id, + session_id=str(session_id), + attachment_id=aid, + source="user_upload", + ) lang = _api_lang(store) apply_gateway_mcp_env_to_os() manager_agent = _init_gateway_executor( @@ -1611,6 +1752,19 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor text = _effective_user_text(text=text_raw, attachments=attachments, store=store) if not _is_administrator_chat_viewer(ctx) and tenant_id and user_id: store.ensure_ui_session_owner(session_id=session_id, tenant_id=tenant_id, user_id=user_id) + if attachments and tenant_id and user_id: + for a in attachments: + if not isinstance(a, dict): + continue + aid = str(a.get("attachment_id") or "").strip().lower() + if aid and _is_valid_attachment_id(aid): + store.link_attachment_acl( + tenant_id=tenant_id, + user_id=user_id, + session_id=str(session_id), + attachment_id=aid, + source="user_upload", + ) lang = _api_lang(store) apply_gateway_mcp_env_to_os() manager_agent = _init_gateway_executor( diff --git a/interfaces/admin/static/chat.html b/interfaces/admin/static/chat.html index 78600cd6..bf99637d 100644 --- a/interfaces/admin/static/chat.html +++ b/interfaces/admin/static/chat.html @@ -130,6 +130,11 @@ .chat-sess-menu-item:hover { background: rgba(255, 255, 255, 0.06); } + .chat-sess-menu-sep { + height: 1px; + margin: 4px 6px; + background: rgba(255, 255, 255, 0.08); + } .chat-msg__md { line-height: 1.45; word-break: break-word; diff --git a/interfaces/admin/static/chat.js b/interfaces/admin/static/chat.js index 6defe772..61058f16 100644 --- a/interfaces/admin/static/chat.js +++ b/interfaces/admin/static/chat.js @@ -148,6 +148,11 @@ const I18N = { "chat.attachment.previewLoading": "加载中…", "chat.attachment.previewError": "预览失败", "chat.attachment.previewEmpty": "(空内容)", + "chat.attachmentAcl": "附件 ACL", + "chat.attachmentAclBackfill": "回填 ACL", + "chat.attachmentAclBackfillPrompt": "回填 attachment_acl(扫描历史消息 attachments)?建议先在低峰期执行。", + "chat.attachmentAclBackfillOk": "回填完成:插入 {inserted} 条(扫描 {scanned_messages} 条消息)", + "chat.attachmentAclBackfillFail": "回填失败:{error}", }, en: { "chat.pageTitle": "oliver", @@ -294,6 +299,11 @@ const I18N = { "chat.attachment.previewLoading": "Loading…", "chat.attachment.previewError": "Preview failed", "chat.attachment.previewEmpty": "(empty)", + "chat.attachmentAcl": "Attachment ACL", + "chat.attachmentAclBackfill": "Backfill ACL", + "chat.attachmentAclBackfillPrompt": "Backfill attachment_acl by scanning historical message attachments? Recommended during off-peak hours.", + "chat.attachmentAclBackfillOk": "Backfill done: inserted {inserted} rows (scanned {scanned_messages} messages)", + "chat.attachmentAclBackfillFail": "Backfill failed: {error}", }, }; @@ -2164,6 +2174,9 @@ function syncAuthUserLabel() { if (!user) return; user.innerHTML = ""; const name = String((authSession && (authSession.display_name || authSession.username || authSession.user_id)) || ""); + const isAdminViewer = String((authSession && authSession.username) || "") + .trim() + .toLowerCase() === "administrator"; if (!name) return; const nameBtn = el("button", { type: "button", @@ -2179,7 +2192,7 @@ function syncAuthUserLabel() { onclick: (ev) => { ev.stopPropagation(); document.querySelectorAll(".chat-sess-menu-pop").forEach((n) => n.remove()); - const menu = el("div", { class: "chat-sess-menu-pop", style: "position:fixed;" }, [ + const items = [ el("button", { type: "button", class: "chat-sess-menu-item", @@ -2198,13 +2211,28 @@ function syncAuthUserLabel() { "data-menu-action": "dispatchLabels", text: t("chat.dispatchLabelsEdit"), }), + ]; + if (isAdminViewer) { + items.push(el("div", { class: "chat-sess-menu-sep" })); + items.push( + el("button", { + type: "button", + class: "chat-sess-menu-item", + "data-menu-action": "attachmentAclBackfill", + text: t("chat.attachmentAclBackfill"), + }), + ); + } + items.push(el("div", { class: "chat-sess-menu-sep" })); + items.push( el("button", { type: "button", class: "chat-sess-menu-item", "data-menu-action": "logout", text: t("auth.logout"), }), - ]); + ); + const menu = el("div", { class: "chat-sess-menu-pop", style: "position:fixed;" }, items); const rect = moreBtn.getBoundingClientRect(); menu.style.left = `${Math.min(rect.left, window.innerWidth - 220)}px`; menu.style.top = `${Math.max(8, rect.top - 92)}px`; @@ -4394,6 +4422,23 @@ document.body.addEventListener("click", async (e) => { await openDispatchLabelsEditor(status); return; } + if (action === "attachmentAclBackfill") { + const status = document.querySelector(".chat-status"); + if (!(await confirmChatAction(t("chat.attachmentAclBackfillPrompt")))) return; + try { + const res = await apiPost("/admin/api/chat/admin/attachments/acl/backfill", {}); + const ok = !!(res && (res.ok === true || res.ok === 1)); + if (ok) { + if (status) status.textContent = t("chat.attachmentAclBackfillOk", res); + } else { + const err = String((res && (res.error || res.detail)) || "backfill_failed"); + if (status) status.textContent = t("chat.attachmentAclBackfillFail", { error: err }); + } + } catch (err) { + if (status) status.textContent = t("chat.attachmentAclBackfillFail", { error: String(err) }); + } + return; + } } }); diff --git a/interfaces/http/weixin_ilink_api.py b/interfaces/http/weixin_ilink_api.py index d23d2847..8b1629d8 100644 --- a/interfaces/http/weixin_ilink_api.py +++ b/interfaces/http/weixin_ilink_api.py @@ -1,11 +1,16 @@ from __future__ import annotations +import asyncio +import os import secrets +import threading import time from typing import Any from fastapi import APIRouter, Header, HTTPException, Request +from oclaw.runtime.application.gateway import process_inbound_payload_usecase + router = APIRouter() @@ -14,7 +19,7 @@ def _require_ilink_auth( *, authorization_type: str | None, authorization: str | None, -) -> None: +) -> str: # Minimal contract required by oclaw-weixin: # - AuthorizationType: ilink_bot_token # - Authorization: Bearer @@ -23,12 +28,259 @@ def _require_ilink_auth( auth = (authorization or "").strip() if not auth.lower().startswith("bearer "): raise HTTPException(status_code=401, detail="missing bearer token") + token = auth[7:].strip() + if not token: + raise HTTPException(status_code=401, detail="empty bearer token") + expected = str(os.getenv("AIA_ILINK_BOT_TOKEN") or "").strip() + if expected and token != expected: + raise HTTPException(status_code=401, detail="invalid bearer token") + return token def _now_ms() -> int: return int(time.time() * 1000) +def _normalize_channel(raw: Any) -> str: + ch = str(raw or "wechat").strip().lower() + return ch if ch else "wechat" + + +def _resolve_account_id(obj: dict[str, Any]) -> str: + for key in ("account_id", "bot_id", "aibotid", "app_id", "agent_id"): + val = obj.get(key) + if val: + return str(val).strip() + return "" + + +def _resolve_stream_selector(body: dict[str, Any]) -> tuple[str, str]: + return _normalize_channel(body.get("channel")), _resolve_account_id(body) + + +def _extract_text_from_body(body: dict[str, Any]) -> str: + direct = body.get("text") + if isinstance(direct, str): + return direct.strip() + content = body.get("content") + if isinstance(content, str): + return content.strip() + if isinstance(content, dict): + text = content.get("text") + if isinstance(text, str): + return text.strip() + msg = body.get("msg") + if isinstance(msg, dict): + text = msg.get("text") + if isinstance(text, str): + return text.strip() + nested = msg.get("content") + if isinstance(nested, str): + return nested.strip() + if isinstance(nested, dict): + text2 = nested.get("text") + if isinstance(text2, str): + return text2.strip() + return "" + + +def _extract_inbound_identity(body: dict[str, Any]) -> tuple[str, str]: + chat_cands = ( + body.get("chat_id"), + body.get("room_id"), + body.get("conversation_id"), + body.get("external_chat_id"), + body.get("to_wxid"), + ) + user_cands = ( + body.get("user_id"), + body.get("external_user_id"), + body.get("from_user"), + body.get("from_wxid"), + body.get("sender"), + body.get("wxid"), + ) + msg = body.get("msg") + if isinstance(msg, dict): + chat_cands = chat_cands + ( + msg.get("chat_id"), + msg.get("room_id"), + msg.get("conversation_id"), + msg.get("to_wxid"), + ) + user_cands = user_cands + ( + msg.get("user_id"), + msg.get("external_user_id"), + msg.get("from_user"), + msg.get("from_wxid"), + msg.get("sender"), + msg.get("wxid"), + ) + user_id = next((str(v).strip() for v in user_cands if str(v or "").strip()), "") + chat_id = next((str(v).strip() for v in chat_cands if str(v or "").strip()), "") + if not chat_id: + chat_id = user_id + return user_id, chat_id + + +class _IlinkBridge: + def __init__(self) -> None: + self._lock = threading.Lock() + self._seq = 0 + self._events: list[dict[str, Any]] = [] + + def enqueue_reply( + self, + *, + token: str, + channel: str, + account_id: str, + chat_id: str, + text: str, + context_token: str | None, + ) -> None: + payload = { + "channel": channel, + "account_id": account_id, + "chat_id": chat_id, + "text": text, + "content": {"type": "text", "text": text}, + # Required by iLink sendmessage protocol; we cache it per reply so the + # bridge can recover across restarts without needing in-memory token cache. + "context_token": (context_token or "").strip(), + "ts": _now_ms(), + } + with self._lock: + self._seq += 1 + event = { + "id": self._seq, + "token": token, + "payload": payload, + } + self._events.append(event) + if len(self._events) > 2000: + self._events = self._events[-1000:] + + def poll( + self, + *, + token: str, + cursor: int, + channel: str, + account_id: str, + limit: int = 50, + ) -> tuple[list[dict[str, Any]], int]: + out: list[dict[str, Any]] = [] + next_cursor = max(0, int(cursor or 0)) + with self._lock: + for event in self._events: + eid = int(event.get("id") or 0) + if eid <= cursor: + continue + if str(event.get("token") or "") != token: + continue + payload = event.get("payload") if isinstance(event.get("payload"), dict) else {} + if channel and str(payload.get("channel") or "") != channel: + continue + if account_id and str(payload.get("account_id") or "") != account_id: + continue + next_cursor = max(next_cursor, eid) + item = dict(payload) + item["id"] = str(eid) + item["msg_id"] = str(eid) + out.append(item) + if len(out) >= max(1, limit): + break + return out, next_cursor + + +_BRIDGE = _IlinkBridge() + + +async def _process_inbound_and_enqueue( + *, + token: str, + channel: str, + account_id: str, + user_id: str, + chat_id: str, + payload: dict[str, Any], +) -> None: + def _extract_context_token(in_payload: dict[str, Any]) -> str: + meta = in_payload.get("metadata") + raw = {} + if isinstance(meta, dict): + raw = meta.get("raw") if isinstance(meta.get("raw"), dict) else {} + if not isinstance(raw, dict): + raw = {} + + # 1) Most direct: raw.msg.context_token (when runner forwards the original msg object). + msg = raw.get("msg") + if isinstance(msg, dict): + ctx = msg.get("context_token") + if ctx is not None: + return str(ctx).strip() + + # 2) Runner also sets raw.metadata.context_token. + raw_meta = raw.get("metadata") + if isinstance(raw_meta, dict): + ctx2 = raw_meta.get("context_token") + if ctx2 is not None: + return str(ctx2).strip() + + # 3) Some clients may put it at top-level. + top_ctx = raw.get("context_token") + if top_ctx is not None: + return str(top_ctx).strip() + + return "" + + try: + out = await asyncio.wait_for(asyncio.to_thread(process_inbound_payload_usecase, payload), timeout=60.0) + except asyncio.TimeoutError: + ctx_token = _extract_context_token(payload) + _BRIDGE.enqueue_reply( + token=token, + channel=channel, + account_id=account_id, + chat_id=chat_id or user_id, + text="系统繁忙,处理超时,请稍后重试。", + context_token=ctx_token, + ) + return + except Exception as exc: + ctx_token = _extract_context_token(payload) + _BRIDGE.enqueue_reply( + token=token, + channel=channel, + account_id=account_id, + chat_id=chat_id or user_id, + text=f"系统错误:{type(exc).__name__}", + context_token=ctx_token, + ) + return + + replies = out.get("replies") if isinstance(out, dict) else [] + if not isinstance(replies, list): + return + for item in replies: + if not isinstance(item, dict): + continue + ctx_token = _extract_context_token(payload) + reply_text = str(item.get("text") or "").strip() + reply_chat_id = str(item.get("chat_id") or chat_id or user_id).strip() or user_id + if not reply_text: + continue + _BRIDGE.enqueue_reply( + token=token, + channel=channel, + account_id=account_id, + chat_id=reply_chat_id, + text=reply_text, + context_token=ctx_token, + ) + + @router.post("/ilink/bot/getupdates") async def ilink_getupdates( body: dict[str, Any], @@ -36,12 +288,52 @@ async def ilink_getupdates( authorizationtype: str | None = Header(default=None, alias="AuthorizationType"), authorization: str | None = Header(default=None, alias="Authorization"), ) -> dict[str, Any]: - _require_ilink_auth(authorization_type=authorizationtype, authorization=authorization) - # TODO: implement real long-poll cursor + message queue. - # For now, return an empty poll result so the plugin can stay connected. + token = _require_ilink_auth(authorization_type=authorizationtype, authorization=authorization) buf = str(body.get("get_updates_buf") or "") + try: + cursor = int(buf or "0") + except Exception: + cursor = 0 + channel, account_id = _resolve_stream_selector(body) + limit_raw = body.get("limit") + try: + limit = max(1, min(int(limit_raw), 100)) + except Exception: + limit = 50 + timeout_raw = body.get("longpolling_timeout_ms") + try: + timeout_ms = max(1000, min(int(timeout_raw), 35_000)) + except Exception: + timeout_ms = 35_000 + deadline = time.time() + (timeout_ms / 1000.0) + msgs: list[dict[str, Any]] = [] + next_cursor = cursor + while time.time() < deadline: + msgs, next_cursor = _BRIDGE.poll( + token=token, + cursor=cursor, + channel=channel, + account_id=account_id, + limit=limit, + ) + if msgs: + break + await asyncio.sleep(0.2) + if not msgs: + msgs, next_cursor = _BRIDGE.poll( + token=token, + cursor=cursor, + channel=channel, + account_id=account_id, + limit=limit, + ) _ = request - return {"ret": 0, "msgs": [], "get_updates_buf": buf, "longpolling_timeout_ms": 35000} + return { + "ret": 0, + "msgs": msgs, + "get_updates_buf": str(next_cursor if msgs else cursor), + "longpolling_timeout_ms": timeout_ms, + } @router.post("/ilink/bot/sendmessage") @@ -50,9 +342,32 @@ async def ilink_sendmessage( authorizationtype: str | None = Header(default=None, alias="AuthorizationType"), authorization: str | None = Header(default=None, alias="Authorization"), ) -> dict[str, Any]: - _require_ilink_auth(authorization_type=authorizationtype, authorization=authorization) - # TODO: map outbound message into our channel sender (when implemented). - _ = body + token = _require_ilink_auth(authorization_type=authorizationtype, authorization=authorization) + channel, account_id = _resolve_stream_selector(body) + user_id, chat_id = _extract_inbound_identity(body) + text = _extract_text_from_body(body) + if not user_id: + return {"ret": 400, "errmsg": "missing user_id"} + if not account_id: + return {"ret": 400, "errmsg": "missing account_id"} + payload = { + "channel": channel, + "account_id": account_id, + "user_id": user_id, + "chat_id": chat_id, + "text": text, + "metadata": {"source": "ilink", "raw": body}, + } + asyncio.create_task( + _process_inbound_and_enqueue( + token=token, + channel=channel, + account_id=account_id, + user_id=user_id, + chat_id=chat_id, + payload=payload, + ) + ) return {"ret": 0} diff --git a/platform/files/attachment_assets.py b/platform/files/attachment_assets.py index 0c7d1534..36e07dee 100644 --- a/platform/files/attachment_assets.py +++ b/platform/files/attachment_assets.py @@ -4,6 +4,7 @@ import base64 import hashlib import json import os +import re import time from dataclasses import dataclass from pathlib import Path @@ -11,6 +12,7 @@ from typing import Any, Optional from oclaw.platform.config.paths import attachments_dir _META_SUFFIX: Final[str] = ".meta.json" +_ATTACHMENT_ID_RE: Final[re.Pattern[str]] = re.compile(r"^[a-f0-9]{64}$") def _utc_ts() -> int: @@ -96,16 +98,23 @@ class AttachmentAssetStore: self.root = Path(root_dir) if root_dir is not None else attachments_dir() self.root.mkdir(parents=True, exist_ok=True) + @staticmethod + def _normalize_attachment_id(attachment_id: str) -> str: + aid = str(attachment_id or "").strip().lower() + if not _ATTACHMENT_ID_RE.fullmatch(aid): + raise ValueError("attachment_id_invalid") + return aid + def _data_path(self, attachment_id: str, *, ext: str) -> Path: # bucket to avoid huge single dir - aid = (attachment_id or "").strip() + aid = self._normalize_attachment_id(attachment_id) p1, p2 = (aid[:2] or "xx"), (aid[2:4] or "yy") d = self.root / p1 / p2 d.mkdir(parents=True, exist_ok=True) return d / f"{aid}{ext}" def _meta_path(self, attachment_id: str) -> Path: - aid = (attachment_id or "").strip() + aid = self._normalize_attachment_id(attachment_id) p1, p2 = (aid[:2] or "xx"), (aid[2:4] or "yy") d = self.root / p1 / p2 d.mkdir(parents=True, exist_ok=True) @@ -158,7 +167,10 @@ class AttachmentAssetStore: return meta def get_meta(self, attachment_id: str) -> Optional[AttachmentMeta]: - mp = self._meta_path(attachment_id) + try: + mp = self._meta_path(attachment_id) + except Exception: + return None if not mp.exists(): return None try: @@ -170,7 +182,10 @@ class AttachmentAssetStore: return None def touch(self, attachment_id: str) -> None: - mp = self._meta_path(attachment_id) + try: + mp = self._meta_path(attachment_id) + except Exception: + return if not mp.exists(): return try: @@ -183,12 +198,16 @@ class AttachmentAssetStore: return def load_bytes(self, attachment_id: str) -> tuple[bytes, Optional[AttachmentMeta]]: - meta = self.get_meta(attachment_id) + try: + aid = self._normalize_attachment_id(attachment_id) + except Exception: + return b"", None + meta = self.get_meta(aid) # try find data file by scanning common extensions exts = (".png", ".jpg", ".jpeg", ".webp", ".gif", "") data_path = None for ext in exts: - p = self._data_path(attachment_id, ext=ext) + p = self._data_path(aid, ext=ext) if p.exists(): data_path = p break @@ -198,15 +217,19 @@ class AttachmentAssetStore: blob = data_path.read_bytes() except Exception: return b"", meta - self.touch(attachment_id) + self.touch(aid) return blob, meta def get_local_path(self, attachment_id: str) -> Path | None: + try: + aid = self._normalize_attachment_id(attachment_id) + except Exception: + return None exts = (".png", ".jpg", ".jpeg", ".webp", ".gif", "") for ext in exts: - p = self._data_path(attachment_id, ext=ext) + p = self._data_path(aid, ext=ext) if p.exists(): - self.touch(attachment_id) + self.touch(aid) return p return None diff --git a/platform/persistence/sqlite_store.py b/platform/persistence/sqlite_store.py index 684160ad..e5c5f16b 100644 --- a/platform/persistence/sqlite_store.py +++ b/platform/persistence/sqlite_store.py @@ -607,6 +607,31 @@ class SqliteStore: """ ) conn.execute("CREATE INDEX IF NOT EXISTS idx_admin_audit_actor_ts ON admin_audit_log(actor_user_id, timestamp DESC)") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS attachment_acl ( + attachment_id TEXT NOT NULL, + tenant_id TEXT NOT NULL, + user_id TEXT NOT NULL, + session_id TEXT NOT NULL, + source TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (attachment_id, tenant_id, user_id, session_id, source), + FOREIGN KEY(session_id) REFERENCES chat_session(id) ON DELETE CASCADE, + FOREIGN KEY(tenant_id) REFERENCES tenant(id) ON DELETE CASCADE, + FOREIGN KEY(user_id) REFERENCES app_user(id) ON DELETE CASCADE + ); + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_attachment_acl_tenant_attachment ON attachment_acl(tenant_id, attachment_id, created_at DESC)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_attachment_acl_user_attachment ON attachment_acl(tenant_id, user_id, attachment_id, created_at DESC)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_attachment_acl_session_attachment ON attachment_acl(session_id, attachment_id, created_at DESC)" + ) conn.execute( """ CREATE TABLE IF NOT EXISTS tool_log ( @@ -1306,6 +1331,206 @@ class SqliteStore: last_message_at=row["last_message_at"], ) + @staticmethod + def _attachments_contain_attachment_id(raw_attachments: Any, *, attachment_id: str) -> bool: + aid = str(attachment_id or "").strip() + if not aid: + return False + obj = raw_attachments + if isinstance(raw_attachments, str): + s = str(raw_attachments or "").strip() + if not s: + return False + try: + obj = json.loads(s) + except Exception: + return False + if isinstance(obj, dict): + items = [obj] + elif isinstance(obj, list): + items = obj + else: + return False + for it in items: + if not isinstance(it, dict): + continue + if str(it.get("attachment_id") or "").strip() == aid: + return True + return False + + def attachment_referenced_by_user(self, *, tenant_id: str, user_id: str, attachment_id: str, scan_limit: int = 2000) -> bool: + aid = str(attachment_id or "").strip() + tid = str(tenant_id or "").strip() + uid = str(user_id or "").strip() + if not aid or not tid or not uid: + return False + with self._connect() as conn: + rows = conn.execute( + """ + SELECT m.attachments + FROM chat_message m + INNER JOIN ui_session_owner o ON o.session_id = m.session_id + WHERE o.tenant_id = ? AND o.user_id = ? AND m.attachments IS NOT NULL AND m.attachments <> '' + ORDER BY m.id DESC + LIMIT ? + """, + (tid, uid, int(max(1, scan_limit))), + ).fetchall() + for r in rows: + if self._attachments_contain_attachment_id(r["attachments"], attachment_id=aid): + return True + return False + + def attachment_referenced_in_tenant(self, *, tenant_id: str, attachment_id: str, scan_limit: int = 4000) -> bool: + aid = str(attachment_id or "").strip() + tid = str(tenant_id or "").strip() + if not aid or not tid: + return False + with self._connect() as conn: + rows = conn.execute( + """ + SELECT m.attachments + FROM chat_message m + INNER JOIN ui_session_owner o ON o.session_id = m.session_id + WHERE o.tenant_id = ? AND m.attachments IS NOT NULL AND m.attachments <> '' + ORDER BY m.id DESC + LIMIT ? + """, + (tid, int(max(1, scan_limit))), + ).fetchall() + for r in rows: + if self._attachments_contain_attachment_id(r["attachments"], attachment_id=aid): + return True + return False + + def link_attachment_acl( + self, + *, + tenant_id: str, + user_id: str, + session_id: str, + attachment_id: str, + source: str, + ) -> None: + tid = str(tenant_id or "").strip() + uid = str(user_id or "").strip() + sid = str(session_id or "").strip() + aid = str(attachment_id or "").strip().lower() + src = str(source or "").strip() or "unknown" + if not tid or not uid or not sid or not aid: + return + ts = utc_now_iso() + with self._connect() as conn: + conn.execute( + """ + INSERT OR IGNORE INTO attachment_acl + (attachment_id, tenant_id, user_id, session_id, source, created_at) + VALUES (?, ?, ?, ?, ?, ?) + """, + (aid, tid, uid, sid, src, ts), + ) + + def attachment_acl_allows_user(self, *, tenant_id: str, user_id: str, attachment_id: str) -> bool: + tid = str(tenant_id or "").strip() + uid = str(user_id or "").strip() + aid = str(attachment_id or "").strip().lower() + if not tid or not uid or not aid: + return False + with self._connect() as conn: + r = conn.execute( + """ + SELECT 1 + FROM attachment_acl + WHERE tenant_id = ? AND user_id = ? AND attachment_id = ? + LIMIT 1 + """, + (tid, uid, aid), + ).fetchone() + return bool(r) + + def attachment_acl_allows_tenant(self, *, tenant_id: str, attachment_id: str) -> bool: + tid = str(tenant_id or "").strip() + aid = str(attachment_id or "").strip().lower() + if not tid or not aid: + return False + with self._connect() as conn: + r = conn.execute( + """ + SELECT 1 + FROM attachment_acl + WHERE tenant_id = ? AND attachment_id = ? + LIMIT 1 + """, + (tid, aid), + ).fetchone() + return bool(r) + + def backfill_attachment_acl_from_messages( + self, + *, + tenant_id: str, + limit_messages: int = 50_000, + ) -> dict[str, Any]: + """Best-effort backfill: scan chat_message.attachments and populate attachment_acl. + + This is intended for one-off migration / operator maintenance. + """ + tid = str(tenant_id or "").strip() + lim = max(1, int(limit_messages)) + if not tid: + return {"ok": False, "error": "tenant_id_required"} + inserted = 0 + scanned_msgs = 0 + scanned_atts = 0 + with self._connect() as conn: + rows = conn.execute( + """ + SELECT m.session_id, m.attachments, o.user_id + FROM chat_message m + INNER JOIN ui_session_owner o ON o.session_id = m.session_id + WHERE o.tenant_id = ? AND m.attachments IS NOT NULL AND m.attachments <> '' + ORDER BY m.id DESC + LIMIT ? + """, + (tid, lim), + ).fetchall() + ts = utc_now_iso() + for r in rows: + scanned_msgs += 1 + sid = str(r["session_id"] or "").strip() + uid = str(r["user_id"] or "").strip() + if not sid or not uid: + continue + try: + obj = json.loads(str(r["attachments"] or "")) + except Exception: + continue + items = obj if isinstance(obj, list) else ([obj] if isinstance(obj, dict) else []) + for a in items: + if not isinstance(a, dict): + continue + scanned_atts += 1 + aid = str(a.get("attachment_id") or "").strip().lower() + if not aid: + continue + src = "backfill:chat_message" + cur = conn.execute( + """ + INSERT OR IGNORE INTO attachment_acl + (attachment_id, tenant_id, user_id, session_id, source, created_at) + VALUES (?, ?, ?, ?, ?, ?) + """, + (aid, tid, uid, sid, src, ts), + ) + inserted += int(cur.rowcount or 0) + return { + "ok": True, + "tenant_id": tid, + "scanned_messages": int(scanned_msgs), + "scanned_attachments": int(scanned_atts), + "inserted": int(inserted), + } + def list_admin_sessions( self, *, diff --git a/runtime/chat/media_redact.py b/runtime/chat/media_redact.py index 3fe4113b..b31362ba 100644 --- a/runtime/chat/media_redact.py +++ b/runtime/chat/media_redact.py @@ -5,6 +5,7 @@ Persistence is untouched; callers use copies when building model context.""" from __future__ import annotations import base64 +import os from typing import Any from oclaw.platform.files.attachment_assets import AttachmentAssetStore @@ -12,6 +13,7 @@ _IMAGE_CONTENT_TYPES = frozenset({"image", "input_image"}) _BASE64_PAYLOAD_KEYS = ("data", "image_base64", "base64", "content_base64", "body_base64") # Below this length we keep values (tiny icons / markers). _MIN_B64_CHARS = 200 +_DEFAULT_MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024 def redact_embedded_image_blobs(obj: Any) -> Any: @@ -89,6 +91,19 @@ def ingest_embedded_image_blobs_as_refs( if raw: blob = _decode_image_bytes(raw) if blob: + max_bytes = _max_attachment_bytes() + if max_bytes > 0 and len(blob) > max_bytes: + redacted = _redact_dict(node) + redacted["type"] = _ref_type_for_mime( + str(node.get("mime") or node.get("mime_type") or "application/octet-stream"), + typ, + ) + redacted["error"] = "attachment_too_large" + redacted["max_bytes"] = int(max_bytes) + redacted["actual_bytes"] = int(len(blob)) + redacted.setdefault("name", str(node.get("name") or "attachment")) + redacted.setdefault("mime", str(node.get("mime") or node.get("mime_type") or "application/octet-stream")) + return redacted idx_seed[0] += 1 mime = str(node.get("mime") or node.get("mime_type") or "image/png").strip() or "image/png" ext = _filename_ext_for_mime(mime) @@ -195,4 +210,14 @@ def _safe_int(raw: Any) -> int | None: return None +def _max_attachment_bytes() -> int: + raw = str(os.getenv("AIA_MAX_ATTACHMENT_BYTES") or "").strip() + if raw.isdigit(): + n = int(raw) + if n <= 0: + return 0 + return min(n, 500 * 1024 * 1024) + return _DEFAULT_MAX_ATTACHMENT_BYTES + + __all__ = ["redact_embedded_image_blobs", "ingest_embedded_image_blobs_as_refs"] diff --git a/runtime/chat/tool_runtime.py b/runtime/chat/tool_runtime.py index f2a36f38..c3b7d748 100644 --- a/runtime/chat/tool_runtime.py +++ b/runtime/chat/tool_runtime.py @@ -889,6 +889,28 @@ class ToolExecutor: event_type="tool_result", event_payload={"tool_name": tc.name, "observed_rows": int(observed_rows_this_call)}, ) + try: + owner = ctx.store.get_ui_session_owner(session_id=ctx.session_id) or {} + tid = str(owner.get("tenant_id") or "").strip() + uid = str(owner.get("user_id") or "").strip() + atts = msg_row.attachments + if tid and uid and atts: + raw = json.loads(atts) if isinstance(atts, str) else atts + items = raw if isinstance(raw, list) else ([raw] if isinstance(raw, dict) else []) + for a in items: + if not isinstance(a, dict): + continue + aid = str(a.get("attachment_id") or "").strip().lower() + if aid: + ctx.store.link_attachment_acl( + tenant_id=tid, + user_id=uid, + session_id=ctx.session_id, + attachment_id=aid, + source=f"tool:{str(tc.name or '')}", + ) + except Exception: + pass tool_msg_write_ms = int((time.perf_counter() - t_db2) * 1000) tool_messages.append({"role": "tool", "tool_call_id": tc.id, "content": tool_content, "name": tc.name}) _trace( diff --git a/runtime/operations/scripts/weixin_install.ps1 b/runtime/operations/scripts/weixin_install.ps1 index e1e4437f..c29270b4 100644 --- a/runtime/operations/scripts/weixin_install.ps1 +++ b/runtime/operations/scripts/weixin_install.ps1 @@ -1,14 +1,15 @@ param( [string]$ChannelId = "oclaw-weixin", - [string]$Package = "@tencent-weixin/oclaw-weixin@2.1.9", - [string]$OclawRuntime = "" + [string]$LocalSourcePath = "", + [switch]$UseOpenclawCli = $false ) $ErrorActionPreference = "Stop" function Resolve-RepoRoot { $here = Split-Path -Parent $PSCommandPath - return (Resolve-Path (Join-Path $here "..")).Path + # runtime/operations/scripts -> repo root + return (Resolve-Path (Join-Path $here "..\\..\\..")).Path } $oclawRoot = Resolve-RepoRoot @@ -19,22 +20,63 @@ New-Item -ItemType Directory -Force -Path $sidecarRoot | Out-Null New-Item -ItemType Directory -Force -Path (Join-Path $sidecarRoot "logs") | Out-Null New-Item -ItemType Directory -Force -Path $stateDir | Out-Null +if ($UseOpenclawCli) { + $openclawCmd = Get-Command openclaw -ErrorAction SilentlyContinue + if (-not $openclawCmd) { + throw "openclaw command not found. Install first: npm install -g openclaw" + } + npx -y @tencent-weixin/openclaw-weixin-cli@latest install + if ($LASTEXITCODE -ne 0) { + throw "openclaw-weixin-cli install failed with exit code $LASTEXITCODE" + } + Push-Location $sidecarRoot + try { + if (-not (Test-Path (Join-Path $sidecarRoot "package.json"))) { + npm.cmd init -y | Out-Null + if ($LASTEXITCODE -ne 0) { + throw "npm init failed with exit code $LASTEXITCODE" + } + } + npm.cmd install --save-exact tsx@4.21.0 typescript@6.0.3 + if ($LASTEXITCODE -ne 0) { + throw "npm install bridge runtime deps failed with exit code $LASTEXITCODE" + } + $bridgeSrc = Join-Path $oclawRoot "runtime\\operations\\weixin_bridge" + Copy-Item -Path (Join-Path $bridgeSrc "runner.ts") -Destination (Join-Path $sidecarRoot "runner.ts") -Force + Copy-Item -Path (Join-Path $bridgeSrc "login.ts") -Destination (Join-Path $sidecarRoot "login.ts") -Force + } finally { + Pop-Location + } + Write-Host "[ok] installed official openclaw-weixin plugin + local bridge runtime" + exit 0 +} + +if (-not $LocalSourcePath) { + throw "LocalSourcePath is required in sidecar mode. Example: .\\scripts\\weixin_install.ps1 -LocalSourcePath D:\\path\\to\\your-weixin-module" +} + Push-Location $sidecarRoot try { if (-not (Test-Path (Join-Path $sidecarRoot "package.json"))) { npm.cmd init -y | Out-Null - } - if (Test-Path (Join-Path $sidecarRoot "package-lock.json")) { - npm.cmd ci - } else { - # First-time setup: install exact versions for reproducible sidecar runtime. - if ($OclawRuntime) { - npm.cmd install --save-exact $Package $OclawRuntime tsx@4.21.0 typescript@6.0.3 - } else { - npm.cmd install --save-exact $Package tsx@4.21.0 typescript@6.0.3 + if ($LASTEXITCODE -ne 0) { + throw "npm init failed with exit code $LASTEXITCODE" } } - Write-Host "[ok] installed $Package into $sidecarRoot" + + $src = (Resolve-Path $LocalSourcePath).Path + npm.cmd install --save-exact $src tsx@4.21.0 typescript@6.0.3 + if ($LASTEXITCODE -ne 0) { + throw "npm install local source failed with exit code $LASTEXITCODE" + } + + if (-not (Test-Path (Join-Path $sidecarRoot "runner.ts"))) { + throw "install completed but runner.ts is missing (invalid sidecar package/source)" + } + if (-not (Test-Path (Join-Path $sidecarRoot "login.ts"))) { + throw "install completed but login.ts is missing (invalid sidecar package/source)" + } + Write-Host "[ok] installed local weixin sidecar into $sidecarRoot" } finally { Pop-Location } diff --git a/runtime/operations/scripts/weixin_login.ps1 b/runtime/operations/scripts/weixin_login.ps1 index 00a62f90..aacbb275 100644 --- a/runtime/operations/scripts/weixin_login.ps1 +++ b/runtime/operations/scripts/weixin_login.ps1 @@ -6,7 +6,8 @@ $ErrorActionPreference = "Stop" function Resolve-RepoRoot { $here = Split-Path -Parent $PSCommandPath - return (Resolve-Path (Join-Path $here "..")).Path + # runtime/operations/scripts -> repo root + return (Resolve-Path (Join-Path $here "..\\..\\..")).Path } $oclawRoot = Resolve-RepoRoot @@ -14,7 +15,7 @@ $sidecarRoot = Join-Path $oclawRoot "data\\channel_sidecar\\$ChannelId" $stateDir = Join-Path $sidecarRoot "state" if (-not (Test-Path $sidecarRoot)) { - throw "sidecar not installed: run .\\scripts\\weixin_install.ps1 first" + New-Item -ItemType Directory -Force -Path $sidecarRoot | Out-Null } New-Item -ItemType Directory -Force -Path $stateDir | Out-Null @@ -22,10 +23,19 @@ New-Item -ItemType Directory -Force -Path $stateDir | Out-Null Push-Location $sidecarRoot try { $env:OCLAW_STATE_DIR = $stateDir - if (-not (Test-Path (Join-Path $sidecarRoot "login.ts"))) { - throw "missing login.ts" + if (Test-Path (Join-Path $sidecarRoot "login.ts")) { + npm.cmd exec -- tsx login.ts + exit 0 } - npm.cmd exec -- tsx login.ts + $openclawCmd = Get-Command openclaw -ErrorAction SilentlyContinue + if (-not $openclawCmd) { + throw "official mode requires openclaw command. Install first: npm install -g openclaw" + } + $systemNodeDir = "C:\\Program Files\\nodejs" + if (Test-Path (Join-Path $systemNodeDir "node.exe")) { + $env:PATH = "$systemNodeDir;$env:PATH" + } + openclaw channels login --channel openclaw-weixin } finally { Pop-Location } diff --git a/runtime/operations/scripts/weixin_start.ps1 b/runtime/operations/scripts/weixin_start.ps1 index 54e4435c..492bb7f1 100644 --- a/runtime/operations/scripts/weixin_start.ps1 +++ b/runtime/operations/scripts/weixin_start.ps1 @@ -7,7 +7,8 @@ $ErrorActionPreference = "Stop" function Resolve-RepoRoot { $here = Split-Path -Parent $PSCommandPath - return (Resolve-Path (Join-Path $here "..")).Path + # runtime/operations/scripts -> repo root + return (Resolve-Path (Join-Path $here "..\\..\\..")).Path } $oclawRoot = Resolve-RepoRoot @@ -16,24 +17,105 @@ $stateDir = Join-Path $sidecarRoot "state" $logDir = Join-Path $sidecarRoot "logs" $pidFile = Join-Path $sidecarRoot "pid.txt" -if (-not (Test-Path $sidecarRoot)) { - throw "sidecar not installed: run .\\scripts\\weixin_install.ps1 first" +function Get-SidecarProcesses { + $escapedSidecarRoot = $sidecarRoot.Replace("\", "\\") + $patterns = @( + "*$ChannelId*", + "*runner.ts*", + "*$escapedSidecarRoot*" + ) + Get-CimInstance Win32_Process | Where-Object { + $cmd = [string]($_.CommandLine) + if (-not $cmd) { return $false } + foreach ($pattern in $patterns) { + if ($cmd -like $pattern) { return $true } + } + return $false + } } -if (-not (Test-Path (Join-Path $sidecarRoot "runner.ts"))) { - throw "missing runner.ts (sidecar code). Re-run repo sync or restore file." + +function Stop-SidecarProcesses { + $procs = @(Get-SidecarProcesses | Sort-Object ProcessId -Descending) + foreach ($proc in $procs) { + try { + taskkill.exe /PID $proc.ProcessId /T /F | Out-Null + } catch { + # Best-effort cleanup; keep going if a process already exited. + } + } + return $procs.Count +} + +function Set-OfficialWeixinBaseUrl([string]$BaseUrl) { + $weixinRoot = Join-Path $env:USERPROFILE ".openclaw\\openclaw-weixin" + $accountsListPath = Join-Path $weixinRoot "accounts.json" + if (-not (Test-Path $accountsListPath)) { + Write-Host "[warn] official mode: accounts.json not found, skip baseUrl rewrite" + return + } + $ids = @() + try { + $parsed = Get-Content -Path $accountsListPath -Raw | ConvertFrom-Json + if ($parsed -is [System.Array]) { + $ids = @($parsed) + } + } catch { + Write-Host "[warn] official mode: failed to parse accounts.json" + return + } + foreach ($aid in $ids) { + $idText = [string]$aid + if (-not $idText) { continue } + $accPath = Join-Path (Join-Path $weixinRoot "accounts") "$idText.json" + if (-not (Test-Path $accPath)) { continue } + try { + $obj = Get-Content -Path $accPath -Raw | ConvertFrom-Json + $obj.baseUrl = $BaseUrl + $json = $obj | ConvertTo-Json -Depth 8 + $utf8NoBom = New-Object System.Text.UTF8Encoding($false) + [System.IO.File]::WriteAllText($accPath, $json + "`n", $utf8NoBom) + Write-Host "[ok] official mode: set baseUrl for $idText -> $BaseUrl" + } catch { + Write-Host "[warn] official mode: failed to rewrite $accPath" + } + } +} + +if (-not (Test-Path $sidecarRoot)) { + New-Item -ItemType Directory -Force -Path $sidecarRoot | Out-Null } New-Item -ItemType Directory -Force -Path $logDir | Out-Null New-Item -ItemType Directory -Force -Path $stateDir | Out-Null +$cleaned = Stop-SidecarProcesses +Remove-Item -Force $pidFile -ErrorAction SilentlyContinue + $logPath = Join-Path $logDir "weixin_sidecar.log" $errPath = Join-Path $logDir "weixin_sidecar.err.log" -$cmd = "cmd.exe" -$args = @( - "/c", - "cd /d $sidecarRoot && set OCLAW_STATE_DIR=$stateDir&& set AIA_GATEWAY_BASE_URL=$GatewayBaseUrl&& npm.cmd exec -- tsx runner.ts" -) +if (Test-Path (Join-Path $sidecarRoot "runner.ts")) { + $cmd = "cmd.exe" + $args = @( + "/c", + "cd /d $sidecarRoot && set OCLAW_STATE_DIR=$stateDir&& set AIA_GATEWAY_BASE_URL=$GatewayBaseUrl&& npm.cmd exec -- tsx runner.ts" + ) + $p = Start-Process -FilePath $cmd -ArgumentList $args -WorkingDirectory $sidecarRoot -PassThru -WindowStyle Hidden -RedirectStandardOutput $logPath -RedirectStandardError $errPath + Set-Content -Path $pidFile -Value $p.Id + Write-Host "[ok] started weixin sidecar pid=$($p.Id) cleaned=$cleaned out=$logPath err=$errPath" + exit 0 +} -$p = Start-Process -FilePath $cmd -ArgumentList $args -WorkingDirectory $sidecarRoot -PassThru -WindowStyle Hidden -RedirectStandardOutput $logPath -RedirectStandardError $errPath +$openclawCmd = Get-Command openclaw -ErrorAction SilentlyContinue +if (-not $openclawCmd) { + throw "official mode requires openclaw command. Install first: npm install -g openclaw" +} +# Ensure OpenClaw runs on the real Node.js runtime (includes npm layout). +$systemNodeDir = "C:\\Program Files\\nodejs" +if (Test-Path (Join-Path $systemNodeDir "node.exe")) { + $env:PATH = "$systemNodeDir;$env:PATH" +} +Set-OfficialWeixinBaseUrl -BaseUrl $GatewayBaseUrl +$args = @("/c", "openclaw gateway --allow-unconfigured") +$p = Start-Process -FilePath "cmd.exe" -ArgumentList $args -WorkingDirectory $oclawRoot -PassThru -WindowStyle Hidden -RedirectStandardOutput $logPath -RedirectStandardError $errPath Set-Content -Path $pidFile -Value $p.Id -Write-Host "[ok] started weixin sidecar pid=$($p.Id) out=$logPath err=$errPath" +Write-Host "[ok] started openclaw gateway bridge pid=$($p.Id) cleaned=$cleaned out=$logPath err=$errPath" diff --git a/runtime/operations/scripts/weixin_status.ps1 b/runtime/operations/scripts/weixin_status.ps1 index 94fd46c2..82f59d24 100644 --- a/runtime/operations/scripts/weixin_status.ps1 +++ b/runtime/operations/scripts/weixin_status.ps1 @@ -6,14 +6,42 @@ $ErrorActionPreference = "Stop" function Resolve-RepoRoot { $here = Split-Path -Parent $PSCommandPath - return (Resolve-Path (Join-Path $here "..")).Path + # runtime/operations/scripts -> repo root + return (Resolve-Path (Join-Path $here "..\\..\\..")).Path } $oclawRoot = Resolve-RepoRoot $sidecarRoot = Join-Path $oclawRoot "data\\channel_sidecar\\$ChannelId" $pidFile = Join-Path $sidecarRoot "pid.txt" +$systemNodeDir = "C:\\Program Files\\nodejs" +if (Test-Path (Join-Path $systemNodeDir "node.exe")) { + $env:PATH = "$systemNodeDir;$env:PATH" +} + +function Get-SidecarProcesses { + $escapedSidecarRoot = $sidecarRoot.Replace("\", "\\") + $patterns = @( + "*$ChannelId*", + "*runner.ts*", + "*$escapedSidecarRoot*" + ) + Get-CimInstance Win32_Process | Where-Object { + $cmd = [string]($_.CommandLine) + if (-not $cmd) { return $false } + foreach ($pattern in $patterns) { + if ($cmd -like $pattern) { return $true } + } + return $false + } +} if (-not (Test-Path $pidFile)) { + $sidecarProcs = @(Get-SidecarProcesses) + if ($sidecarProcs.Count -gt 0) { + $pids = ($sidecarProcs | Select-Object -ExpandProperty ProcessId) -join "," + Write-Host "status=orphaned count=$($sidecarProcs.Count) pids=$pids" + exit 0 + } Write-Host "status=stopped" exit 0 } @@ -31,8 +59,25 @@ try { } catch {} if ($exists) { - Write-Host "status=running pid=$procId" + $sidecarProcs = @(Get-SidecarProcesses) + Write-Host "status=running pid=$procId matches=$($sidecarProcs.Count)" } else { + $sidecarProcs = @(Get-SidecarProcesses) + if ($sidecarProcs.Count -gt 0) { + $pids = ($sidecarProcs | Select-Object -ExpandProperty ProcessId) -join "," + Write-Host "status=orphaned stale_pid=$procId count=$($sidecarProcs.Count) pids=$pids" + exit 0 + } + $openclawCmd = Get-Command openclaw -ErrorAction SilentlyContinue + if ($openclawCmd) { + try { + $txt = (& openclaw channels status --probe) -join "`n" + if ($txt -match "openclaw-weixin .*running") { + Write-Host "status=running mode=official" + exit 0 + } + } catch {} + } Write-Host "status=stale_pid pid=$procId" } diff --git a/runtime/operations/scripts/weixin_stop.ps1 b/runtime/operations/scripts/weixin_stop.ps1 index bc565e73..a998b899 100644 --- a/runtime/operations/scripts/weixin_stop.ps1 +++ b/runtime/operations/scripts/weixin_stop.ps1 @@ -4,17 +4,63 @@ param( ) $ErrorActionPreference = "Stop" +$systemNodeDir = "C:\\Program Files\\nodejs" +if (Test-Path (Join-Path $systemNodeDir "node.exe")) { + $env:PATH = "$systemNodeDir;$env:PATH" +} function Resolve-RepoRoot { $here = Split-Path -Parent $PSCommandPath - return (Resolve-Path (Join-Path $here "..")).Path + # runtime/operations/scripts -> repo root + return (Resolve-Path (Join-Path $here "..\\..\\..")).Path } $oclawRoot = Resolve-RepoRoot $sidecarRoot = Join-Path $oclawRoot "data\\channel_sidecar\\$ChannelId" $pidFile = Join-Path $sidecarRoot "pid.txt" +function Get-SidecarProcesses { + $escapedSidecarRoot = $sidecarRoot.Replace("\", "\\") + $patterns = @( + "*$ChannelId*", + "*runner.ts*", + "*$escapedSidecarRoot*" + ) + Get-CimInstance Win32_Process | Where-Object { + $cmd = [string]($_.CommandLine) + if (-not $cmd) { return $false } + foreach ($pattern in $patterns) { + if ($cmd -like $pattern) { return $true } + } + return $false + } +} + +function Stop-SidecarProcesses { + param( + [switch]$ForceKill + ) + $procs = @(Get-SidecarProcesses | Sort-Object ProcessId -Descending) + foreach ($proc in $procs) { + try { + if ($ForceKill) { + taskkill.exe /PID $proc.ProcessId /T /F | Out-Null + } else { + taskkill.exe /PID $proc.ProcessId /T | Out-Null + } + } catch { + # Ignore already-exited processes and continue best-effort cleanup. + } + } + return $procs.Count +} + if (-not (Test-Path $pidFile)) { + $killed = Stop-SidecarProcesses -ForceKill:$Force + if ($killed -gt 0) { + Write-Host "[ok] cleaned stale sidecar processes count=$killed" + exit 0 + } Write-Host "[ok] not running (no pid file)" exit 0 } @@ -36,6 +82,11 @@ try { # Ignore if already dead. } +$killed = Stop-SidecarProcesses -ForceKill:$Force Remove-Item -Force $pidFile -ErrorAction SilentlyContinue -Write-Host "[ok] stopped pid=$procId" +$openclawCmd = Get-Command openclaw -ErrorAction SilentlyContinue +if ($openclawCmd) { + try { openclaw gateway stop | Out-Null } catch {} +} +Write-Host "[ok] stopped pid=$procId extra_cleaned=$killed" diff --git a/runtime/operations/weixin_bridge/login.ts b/runtime/operations/weixin_bridge/login.ts new file mode 100644 index 00000000..e6c9a333 --- /dev/null +++ b/runtime/operations/weixin_bridge/login.ts @@ -0,0 +1,16 @@ +import { spawn } from "node:child_process"; + +function run(): Promise { + return new Promise((resolve, reject) => { + const child = spawn("openclaw", ["channels", "login", "--channel", "openclaw-weixin"], { + stdio: "inherit", + shell: true, + }); + child.on("error", reject); + child.on("exit", (code) => resolve(code ?? 1)); + }); +} + +void run().then((code) => { + process.exitCode = code; +}); diff --git a/runtime/operations/weixin_bridge/runner.ts b/runtime/operations/weixin_bridge/runner.ts new file mode 100644 index 00000000..f02ab158 --- /dev/null +++ b/runtime/operations/weixin_bridge/runner.ts @@ -0,0 +1,396 @@ +import fs from "node:fs"; +import crypto from "node:crypto"; +import os from "node:os"; +import path from "node:path"; + +type Json = Record; +type TokenMap = Record; + +const LOCAL_BASE_URL = (process.env.AIA_GATEWAY_BASE_URL || "http://127.0.0.1:8787").trim(); +const STATE_DIR = (process.env.OCLAW_STATE_DIR || path.resolve(process.cwd(), "state")).trim(); +const STATE_FILE = path.join(STATE_DIR, "bridge_state.json"); +const POLL_TIMEOUT_MS = 5000; +const CHANNEL_VERSION = "2.1.10"; +const ILINK_APP_ID = "bot"; +const ILINK_APP_CLIENT_VERSION = "131338"; + +class RequestTimeoutError extends Error { + endpoint: string; + + constructor(endpoint: string, timeoutMs: number) { + super(`timeout endpoint=${endpoint} timeoutMs=${timeoutMs}`); + this.name = "RequestTimeoutError"; + this.endpoint = endpoint; + } +} + +function log(msg: string): void { + const ts = new Date().toISOString(); + process.stdout.write(`${ts} [bridge] ${msg}\n`); +} + +function ensureDir(dir: string): void { + fs.mkdirSync(dir, { recursive: true }); +} + +function readJsonFile(p: string): T | null { + try { + return JSON.parse(fs.readFileSync(p, "utf8")) as T; + } catch { + return null; + } +} + +function writeJsonFile(p: string, obj: unknown): void { + fs.writeFileSync(p, JSON.stringify(obj, null, 2) + "\n", "utf8"); +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function homeOpenclawPath(...parts: string[]): string { + return path.join(os.homedir(), ".openclaw", ...parts); +} + +function resolveAccount(): { accountId: string; token: string; cloudBaseUrl: string } { + const ids = readJsonFile(homeOpenclawPath("openclaw-weixin", "accounts.json")) || []; + const accountId = String(ids[0] || "").trim(); + if (!accountId) { + throw new Error("no weixin account id found; run login first"); + } + const account = readJsonFile(homeOpenclawPath("openclaw-weixin", "accounts", `${accountId}.json`)) || {}; + const token = String(account.token || "").trim(); + if (!token) { + throw new Error(`missing token for account ${accountId}; run login again`); + } + const envCloud = String(process.env.OCLAW_WEIXIN_CLOUD_BASE_URL || "").trim(); + const cfgCloud = String(account.baseUrl || "").trim(); + const cloudBaseUrl = (envCloud || cfgCloud || "https://ilinkai.weixin.qq.com").trim(); + const low = cloudBaseUrl.toLowerCase(); + if ( + low.startsWith("http://127.0.0.1") + || low.startsWith("http://localhost") + || low.startsWith("https://127.0.0.1") + || low.startsWith("https://localhost") + ) { + throw new Error( + `invalid cloud baseUrl (${cloudBaseUrl}). It looks like the account file was overwritten. ` + + `Re-run: openclaw channels login --channel openclaw-weixin (QR scan) to restore the cloud baseUrl.`, + ); + } + return { accountId, token, cloudBaseUrl }; +} + +async function postJson(baseUrl: string, endpoint: string, body: Json, token: string, timeoutMs: number): Promise { + const url = `${baseUrl.replace(/\/+$/, "")}/${endpoint.replace(/^\/+/, "")}`; + const ctl = new AbortController(); + const timer = setTimeout(() => ctl.abort(), timeoutMs); + const wrapped: Json = { ...body, base_info: { channel_version: CHANNEL_VERSION } }; + const uin = Buffer.from(String(Math.floor(Math.random() * 0xffffffff)), "utf-8").toString("base64"); + try { + let res: Response; + try { + res = await fetch(url, { + method: "POST", + headers: { + "Content-Type": "application/json", + AuthorizationType: "ilink_bot_token", + Authorization: `Bearer ${token}`, + "X-WECHAT-UIN": uin, + "iLink-App-Id": ILINK_APP_ID, + "iLink-App-ClientVersion": ILINK_APP_CLIENT_VERSION, + }, + body: JSON.stringify(wrapped), + signal: ctl.signal, + }); + } catch (err) { + if (err instanceof Error && err.name === "AbortError") { + throw new RequestTimeoutError(endpoint, timeoutMs); + } + throw err; + } + const text = await res.text(); + if (!res.ok) { + throw new Error(`HTTP ${res.status} ${endpoint}: ${text.slice(0, 300)}`); + } + const parsed = text ? (JSON.parse(text) as Json) : {}; + if (Object.prototype.hasOwnProperty.call(parsed, "ret")) { + const ret = Number((parsed as Json).ret ?? 0); + if (Number.isFinite(ret) && ret !== 0) { + const errcode = (parsed as Json).errcode; + const errmsg = String((parsed as Json).errmsg || ""); + throw new Error(`ret=${ret} errcode=${String(errcode ?? "")} errmsg=${errmsg} endpoint=${endpoint}`); + } + } + return parsed; + } finally { + clearTimeout(timer); + } +} + +function extractTextItems(msg: Json): string { + const list = Array.isArray(msg.item_list) ? msg.item_list : []; + const parts: string[] = []; + for (const item of list) { + if (!item || typeof item !== "object") continue; + const row = item as Json; + const t = Number(row.type || 0); + if (t !== 1) continue; + const textItem = row.text_item; + if (textItem && typeof textItem === "object") { + const val = String((textItem as Json).text || "").trim(); + if (val) parts.push(val); + } + } + return parts.join("\n").trim(); +} + +function toNumber(v: unknown, fallback = 0): number { + const n = Number(v); + return Number.isFinite(n) ? n : fallback; +} + +function generateClientId(): string { + // Match the official openclaw-weixin plugin behavior (util/random.ts). + // Format: `{prefix}:{timestamp}-{8-char hex}` + return `openclaw-weixin:${Date.now()}-${crypto.randomBytes(4).toString("hex")}`; +} + +function parseCursor(v: string): number { + const n = Number(v); + return Number.isFinite(n) ? Math.max(0, Math.floor(n)) : 0; +} + +async function normalizeLocalCursor(args: { + token: string; + accountId: string; + localCursor: string; +}): Promise { + const cursor = String(args.localCursor || "").trim(); + const currentInt = parseCursor(cursor); + if (!cursor || currentInt <= 0) return cursor; + + const current = await postJson( + LOCAL_BASE_URL, + "ilink/bot/getupdates", + { + channel: "wechat", + account_id: args.accountId, + get_updates_buf: cursor, + longpolling_timeout_ms: 1000, + limit: 1, + }, + args.token, + 8000, + ); + const currentMsgs = Array.isArray(current.msgs) ? current.msgs : []; + if (currentMsgs.length > 0) { + return cursor; + } + + const probe = await postJson( + LOCAL_BASE_URL, + "ilink/bot/getupdates", + { + channel: "wechat", + account_id: args.accountId, + get_updates_buf: "0", + longpolling_timeout_ms: 1000, + limit: 1, + }, + args.token, + 8000, + ); + const probeMsgs = Array.isArray(probe.msgs) ? probe.msgs : []; + const probeCursor = parseCursor(String(probe.get_updates_buf || "0")); + if (probeMsgs.length > 0 && probeCursor <= currentInt) { + log(`local cursor looks stale after gateway restart; reset ${cursor} -> 0`); + return "0"; + } + return cursor; +} + +async function forwardInboundToLocal(args: { + token: string; + accountId: string; + msg: Json; +}): Promise { + const fromUser = String(args.msg.from_user_id || "").trim(); + const toUser = String(args.msg.to_user_id || "").trim(); + if (!fromUser) return; + if (toUser && fromUser === toUser) return; + if (toNumber(args.msg.message_type, 1) !== 1) return; // only user -> bot + const text = extractTextItems(args.msg); + if (!text) return; + const contextToken = String(args.msg.context_token || "").trim(); + await postJson( + LOCAL_BASE_URL, + "ilink/bot/sendmessage", + { + channel: "wechat", + account_id: args.accountId, + user_id: fromUser, + chat_id: fromUser, + text, + msg: args.msg, + metadata: { + context_token: contextToken, + }, + }, + args.token, + 15000, + ); + log(`inbound forwarded: from=${fromUser} textLen=${text.length}`); +} + +async function flushLocalReplies(args: { + token: string; + accountId: string; + localCursor: string; + userContextTokens: TokenMap; + cloudBaseUrl: string; +}): Promise { + let cursor = args.localCursor; + for (let i = 0; i < 3; i += 1) { + const out = await postJson( + LOCAL_BASE_URL, + "ilink/bot/getupdates", + { + channel: "wechat", + account_id: args.accountId, + get_updates_buf: cursor, + longpolling_timeout_ms: 1000, + limit: 20, + }, + args.token, + 8000, + ); + const msgs = Array.isArray(out.msgs) ? (out.msgs as Json[]) : []; + const next = String(out.get_updates_buf || cursor || "").trim(); + const batchCursor = cursor; + const nextCursor = next || cursor; + if (!msgs.length) { + break; + } + let allSucceeded = true; + for (const r of msgs) { + const toUser = String(r.chat_id || "").trim(); + const text = String(r.text || "").trim(); + if (!toUser || !text) continue; + const contextToken = String( + (r.context_token as string) || args.userContextTokens[toUser] || "", + ).trim(); + if (!contextToken) { + // Don't advance cursor when we can't produce a valid protocol reply. + allSucceeded = false; + log(`reply missing context_token; keep cursor. to=${toUser} textLen=${text.length}`); + continue; + } + const msgBody: Json = { + from_user_id: "", + to_user_id: toUser, + client_id: generateClientId(), + message_type: 2, + message_state: 2, + item_list: [{ type: 1, text_item: { text } }], + context_token: contextToken || undefined, + }; + try { + await postJson( + args.cloudBaseUrl, + "ilink/bot/sendmessage", + { + msg: msgBody, + }, + args.token, + 12000, + ); + log(`reply pushed: to=${toUser} textLen=${text.length}`); + } catch (err) { + log(`reply push failed: to=${toUser} err=${String(err)}`); + allSucceeded = false; + } + } + cursor = allSucceeded ? nextCursor : batchCursor; + } + return cursor; +} + +async function main(): Promise { + ensureDir(STATE_DIR); + const state = (readJsonFile(STATE_FILE) || {}) as Json; + let cloudCursor = String(state.cloud_cursor || "").trim(); + let localCursor = String(state.local_cursor || "").trim(); + const userContextTokens: TokenMap = + typeof state.user_context_tokens === "object" && state.user_context_tokens + ? (state.user_context_tokens as TokenMap) + : {}; + const { accountId, token, cloudBaseUrl } = resolveAccount(); + localCursor = await normalizeLocalCursor({ token, accountId, localCursor }); + log(`bridge started account=${accountId} cloud=${cloudBaseUrl} local=${LOCAL_BASE_URL}`); + // Ensure local path is healthy before entering long poll. + await postJson( + LOCAL_BASE_URL, + "ilink/bot/getupdates", + { + channel: "wechat", + account_id: accountId, + get_updates_buf: localCursor, + longpolling_timeout_ms: 1000, + }, + token, + 8000, + ); + while (true) { + try { + localCursor = await flushLocalReplies({ + token, + accountId, + localCursor, + userContextTokens, + cloudBaseUrl, + }); + const out = await postJson( + cloudBaseUrl, + "ilink/bot/getupdates", + { + get_updates_buf: cloudCursor, + longpolling_timeout_ms: POLL_TIMEOUT_MS, + }, + token, + POLL_TIMEOUT_MS + 5000, + ); + const msgs = Array.isArray(out.msgs) ? (out.msgs as Json[]) : []; + const nextCloudCursor = String(out.get_updates_buf || cloudCursor || "").trim(); + if (nextCloudCursor) cloudCursor = nextCloudCursor; + for (const msg of msgs) { + const fromUser = String(msg.from_user_id || "").trim(); + const contextToken = String(msg.context_token || "").trim(); + if (fromUser && contextToken) userContextTokens[fromUser] = contextToken; + await forwardInboundToLocal({ token, accountId, msg }); + } + localCursor = await flushLocalReplies({ + token, + accountId, + localCursor, + userContextTokens, + cloudBaseUrl, + }); + writeJsonFile(STATE_FILE, { + cloud_cursor: cloudCursor, + local_cursor: localCursor, + user_context_tokens: userContextTokens, + updated_at: new Date().toISOString(), + }); + } catch (err) { + if (err instanceof RequestTimeoutError && err.endpoint === "ilink/bot/getupdates") { + continue; + } + log(`loop error: ${String(err)}`); + await sleep(1200); + } + } +} + +void main(); diff --git a/tests/test_attachment_acl_backfill.py b/tests/test_attachment_acl_backfill.py new file mode 100644 index 00000000..52be39ce --- /dev/null +++ b/tests/test_attachment_acl_backfill.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import base64 +import hashlib +import os +import tempfile +import unittest +from pathlib import Path + +from fastapi.testclient import TestClient + +from oclaw.interfaces.http.fastapi_app import create_app +from oclaw.platform.files.attachment_assets import AttachmentAssetStore +from oclaw.platform.persistence.sqlite_store import SqliteStore + + +def _pw_hash(pw: str) -> str: + return hashlib.sha256(pw.encode("utf-8")).hexdigest() + + +class AttachmentAclBackfillTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.db = Path(self._tmp.name) / "ops.sqlite" + os.environ["OPS_ASSISTANT_DB_PATH"] = str(self.db) + os.environ["OPS_ASSISTANT_PASSWORD"] = "test-admin-pass" + store = SqliteStore(str(self.db)) + t = store.create_tenant("Team") + self.tenant_id = str(t["id"]) + store.create_user_account( + tenant_id=self.tenant_id, + username="administrator", + display_name="Admin", + role="owner", + password_hash=_pw_hash("test-admin-pass"), + is_active=True, + ) + store.create_user_account( + tenant_id=self.tenant_id, + username="alice", + display_name="Alice", + role="member", + password_hash=_pw_hash("alice-pass"), + is_active=True, + ) + self.alice_id = str(store.get_user_by_username(tenant_id=self.tenant_id, username="alice")["id"]) + self.client = TestClient(create_app()) + self.client.post("/admin/api/auth/bootstrap", json={}) + tok = self.client.post( + "/admin/api/auth/login", + json={"tenant_id": self.tenant_id, "username": "alice", "password": "alice-pass", "purpose": "chat"}, + ).json() + self.token = str(tok.get("token") or "") + + def tearDown(self) -> None: + self._tmp.cleanup() + + def _h(self) -> dict[str, str]: + return {"authorization": f"Bearer {self.token}"} + + def test_backfill_enables_strict_acl_download(self) -> None: + store = SqliteStore(str(self.db)) + sess = store.create_session_for_user(title="t", tenant_id=self.tenant_id, user_id=self.alice_id) + ast = AttachmentAssetStore() + blob = base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==") + meta = ast.save_bytes(blob, filename="x.png", mime="image/png") + aid = str(meta.attachment_id or "").strip() + store.add_message( + session_id=str(sess.id), + role="assistant", + content="here", + attachments=[{"type": "image_ref", "attachment_id": aid, "mime": "image/png", "name": "x.png"}], + ) + + prev = os.environ.get("AIA_ATTACHMENT_ACL_STRICT") + os.environ["AIA_ATTACHMENT_ACL_STRICT"] = "1" + try: + # Strict mode: without backfill, this should be forbidden (not avatar). + r0 = self.client.get(f"/admin/api/chat/attachments/{aid}", headers=self._h()) + self.assertEqual(r0.status_code, 403, r0.text) + + # Backfill ACL for tenant, then strict download should succeed. + res = store.backfill_attachment_acl_from_messages(tenant_id=self.tenant_id, limit_messages=5000) + self.assertTrue(res.get("ok"), res) + r1 = self.client.get(f"/admin/api/chat/attachments/{aid}", headers=self._h()) + self.assertEqual(r1.status_code, 200, r1.text) + self.assertTrue(len(r1.content) > 10) + finally: + if prev is None: + os.environ.pop("AIA_ATTACHMENT_ACL_STRICT", None) + else: + os.environ["AIA_ATTACHMENT_ACL_STRICT"] = prev + + +if __name__ == "__main__": + unittest.main() + diff --git a/tests/test_chat_profile_api.py b/tests/test_chat_profile_api.py index f73ad102..d514d400 100644 --- a/tests/test_chat_profile_api.py +++ b/tests/test_chat_profile_api.py @@ -47,6 +47,25 @@ class ChatProfileApiTests(unittest.TestCase): ) self.assertEqual(resp.status_code, 200, resp.text) self.token = str(resp.json().get("token") or "") + store.create_user_account( + tenant_id=self.tenant_id, + username="alice", + display_name="Alice", + role="member", + password_hash=hashlib.sha256("alice-pass".encode("utf-8")).hexdigest(), + is_active=True, + ) + resp_alice = self.client.post( + "/admin/api/auth/login", + json={ + "tenant_id": self.tenant_id, + "username": "alice", + "password": "alice-pass", + "purpose": "console", + }, + ) + self.assertEqual(resp_alice.status_code, 200, resp_alice.text) + self.alice_token = str(resp_alice.json().get("token") or "") def tearDown(self) -> None: self._tmp.cleanup() @@ -54,6 +73,9 @@ class ChatProfileApiTests(unittest.TestCase): def _h(self) -> dict[str, str]: return {"authorization": f"Bearer {self.token}"} + def _h_alice(self) -> dict[str, str]: + return {"authorization": f"Bearer {self.alice_token}"} + def test_profile_get_patch_avatar_delete(self) -> None: g = self.client.get("/admin/api/chat/profile", headers=self._h()) self.assertEqual(g.status_code, 200, g.text) @@ -92,6 +114,50 @@ class ChatProfileApiTests(unittest.TestCase): g3 = self.client.get("/admin/api/chat/profile", headers=self._h()) self.assertFalse((g3.json().get("profile") or {}).get("avatar_attachment_id")) + def test_attachment_endpoint_rejects_invalid_attachment_id(self) -> None: + r = self.client.get("/admin/api/chat/attachments/not-a-valid-id", headers=self._h()) + self.assertEqual(r.status_code, 400, r.text) + self.assertEqual((r.json() or {}).get("detail"), "attachment_id_invalid") + + def test_attachment_endpoint_forbids_unowned_attachment(self) -> None: + up = self.client.post( + "/admin/api/chat/profile/avatar", + files={"file": ("x.png", MINI_PNG, "image/png")}, + headers=self._h(), + ) + self.assertEqual(up.status_code, 200, up.text) + aid = str((up.json() or {}).get("avatar_attachment_id") or "").strip() + self.assertTrue(aid) + + me = self.client.get(f"/admin/api/chat/attachments/{aid}", headers=self._h()) + self.assertEqual(me.status_code, 200, me.text) + + other = self.client.get(f"/admin/api/chat/attachments/{aid}", headers=self._h_alice()) + self.assertEqual(other.status_code, 403, other.text) + self.assertEqual((other.json() or {}).get("detail"), "attachment_forbidden") + + def test_attachment_acl_strict_requires_backfill_or_acl(self) -> None: + prev = os.environ.get("AIA_ATTACHMENT_ACL_STRICT") + os.environ["AIA_ATTACHMENT_ACL_STRICT"] = "1" + try: + # Upload an avatar (this is not linked into ACL; access is via avatar_attachment_id). + up = self.client.post( + "/admin/api/chat/profile/avatar", + files={"file": ("x.png", MINI_PNG, "image/png")}, + headers=self._h(), + ) + self.assertEqual(up.status_code, 200, up.text) + aid = str((up.json() or {}).get("avatar_attachment_id") or "").strip() + self.assertTrue(aid) + # Avatar download remains allowed under strict mode. + att = self.client.get(f"/admin/api/chat/attachments/{aid}", headers=self._h()) + self.assertEqual(att.status_code, 200, att.text) + finally: + if prev is None: + os.environ.pop("AIA_ATTACHMENT_ACL_STRICT", None) + else: + os.environ["AIA_ATTACHMENT_ACL_STRICT"] = prev + if __name__ == "__main__": unittest.main() diff --git a/tests/test_media_redact.py b/tests/test_media_redact.py index e8698ded..bb1e90f5 100644 --- a/tests/test_media_redact.py +++ b/tests/test_media_redact.py @@ -1,6 +1,7 @@ from __future__ import annotations import base64 +import os from oclaw.runtime.chat.media_redact import ingest_embedded_image_blobs_as_refs, redact_embedded_image_blobs @@ -42,3 +43,22 @@ def test_ingest_embedded_binary_blob_as_binary_ref(tmp_path) -> None: assert str(block.get("attachment_id") or "") assert "base64" not in block assert refs and refs[0]["attachment_id"] == block["attachment_id"] + + +def test_ingest_rejects_oversized_blob_by_env_limit(tmp_path) -> None: + prev = os.environ.get("AIA_MAX_ATTACHMENT_BYTES") + os.environ["AIA_MAX_ATTACHMENT_BYTES"] = "4" + try: + raw = base64.b64encode(b"12345").decode("ascii") + obj = {"result": {"content": [{"type": "file", "mime": "application/octet-stream", "base64": raw, "name": "x.bin"}]}} + out, refs = ingest_embedded_image_blobs_as_refs(obj, root_dir=str(tmp_path), filename_prefix="unit") + finally: + if prev is None: + os.environ.pop("AIA_MAX_ATTACHMENT_BYTES", None) + else: + os.environ["AIA_MAX_ATTACHMENT_BYTES"] = prev + block = out["result"]["content"][0] + assert block["type"] == "binary_ref" + assert block.get("error") == "attachment_too_large" + assert int(block.get("actual_bytes") or 0) >= 5 + assert refs == [] diff --git a/tests/test_weixin_ilink_api.py b/tests/test_weixin_ilink_api.py new file mode 100644 index 00000000..1cdd2183 --- /dev/null +++ b/tests/test_weixin_ilink_api.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import unittest + +from fastapi.testclient import TestClient + +from oclaw.interfaces.http.fastapi_app import create_app +from oclaw.interfaces.http import weixin_ilink_api + + +class WeixinIlinkApiTests(unittest.TestCase): + def setUp(self) -> None: + self.client = TestClient(create_app()) + self.headers = { + "AuthorizationType": "ilink_bot_token", + "Authorization": "Bearer test-ilink-token", + } + + def test_sendmessage_requires_account_id(self) -> None: + r = self.client.post( + "/ilink/bot/sendmessage", + headers=self.headers, + json={ + "channel": "wechat", + "user_id": "wxid_u1", + "text": "hello", + }, + ) + self.assertEqual(r.status_code, 200, r.text) + self.assertEqual((r.json() or {}).get("ret"), 400) + + def test_sendmessage_enqueues_reply_for_getupdates(self) -> None: + old_usecase = weixin_ilink_api.process_inbound_payload_usecase + + def _fake_usecase(payload: dict[str, object]) -> dict[str, object]: + text = str(payload.get("text") or "") + return { + "ok": True, + "replies": [ + { + "chat_id": str(payload.get("chat_id") or ""), + "text": f"echo:{text}", + } + ], + } + + try: + weixin_ilink_api.process_inbound_payload_usecase = _fake_usecase # type: ignore[assignment] + s = self.client.post( + "/ilink/bot/sendmessage", + headers=self.headers, + json={ + "channel": "wechat", + "account_id": "bot-1", + "user_id": "wxid_u2", + "chat_id": "room_1", + "text": "ping", + }, + ) + self.assertEqual(s.status_code, 200, s.text) + self.assertEqual((s.json() or {}).get("ret"), 0) + + g = self.client.post( + "/ilink/bot/getupdates", + headers=self.headers, + json={ + "channel": "wechat", + "account_id": "bot-1", + "get_updates_buf": "0", + "longpolling_timeout_ms": 1000, + }, + ) + self.assertEqual(g.status_code, 200, g.text) + data = g.json() or {} + self.assertEqual(data.get("ret"), 0) + msgs = data.get("msgs") if isinstance(data.get("msgs"), list) else [] + self.assertTrue(msgs, data) + self.assertEqual(str((msgs[0] or {}).get("text") or ""), "echo:ping") + finally: + weixin_ilink_api.process_inbound_payload_usecase = old_usecase # type: ignore[assignment] + + +if __name__ == "__main__": + unittest.main()