diff --git a/docs/multi-tenant/cloud-v2-pending-verification.md b/docs/multi-tenant/cloud-v2-pending-verification.md index 06ffa02aa..908005ba4 100644 --- a/docs/multi-tenant/cloud-v2-pending-verification.md +++ b/docs/multi-tenant/cloud-v2-pending-verification.md @@ -5,7 +5,9 @@ 更新日期：2026-07-29 本文是 Cloud v2 首期上线前的剩余验证清单。它只记录尚不能由当前代码审查、 -单元测试、集成测试、合成容量探针或短时 Linux 容器实验替代的证据。 +单元测试、集成测试、合成容量探针或短时 Linux 容器实验替代的证据。这里的项目 +不属于 2026-07-29 代码与本地测试资源审查的完成条件，也不会让该审查持续保持未完成； +它们只在准备最终 SaaS 激活时重新进入验收范围。 相关文档： @@ -16,8 +18,8 @@ ## 1. 当前已形成的交付基线 -- LangBot Core 全量 `2839 passed, 33 skipped`，Plugin SDK 全量 - `1325 passed`，闭源适配器 `40 passed`，Space Go 全量测试通过；三仓格式、 +- LangBot Core 全量 `2855 passed, 33 skipped`，Plugin SDK 全量 + `1328 passed`，闭源适配器 `40 passed`，Space Go 全量测试通过；三仓格式、 静态检查和 `git diff --check` 已通过。 - Plugin Runtime 和 Box Runtime 的公开健康接口、event-loop lag 与有界 blocking executor 指标已经过真实进程短时验证。 @@ -26,7 +28,7 @@ - PostgreSQL 16 + RLS 的 1,000 Workspace 真实启动测试，以及 5,000 Workspace 三代替换合成探针已通过。 - Core 已精确钉住 Plugin SDK 提交 - `7c0b9827ed8597a1c84151b83fcf6307934fd944`。最终验证必须使用包含该提交的 + `87d903e9b7c4e518ce066c89f65bbde4de825fc5`。最终验证必须使用包含该提交的 Core、Plugin Runtime 和 Box Runtime 镜像，不能混用旧 SDK。 - 独立资源复核已经移除 Cloud MCP 每会话 5 秒查询执行绑定的轮询，改由签名目录 投影提交后向一个合并回收任务发布代次变化；工具与资源调用前后仍使用数据库 diff --git a/docs/multi-tenant/runtime-resource-audit-2026-07-28.md b/docs/multi-tenant/runtime-resource-audit-2026-07-28.md index 3c1411339..81a59229a 100644 --- a/docs/multi-tenant/runtime-resource-audit-2026-07-28.md +++ b/docs/multi-tenant/runtime-resource-audit-2026-07-28.md @@ -9,15 +9,15 @@ ## 结论 -本轮已覆盖 LangBot Core、Plugin Runtime 和 Box Runtime 的主要常驻对象、后台任务、队列、网络客户端、进程生命周期及数据库连接池。本轮定位到的攻击者可控或历史累积状态均已补充容量、超时、淘汰或确定性清理边界；修正后的高基数探针没有观察到随历史请求继续增长的活跃缓存。该结论不等于证明任意生产负载下不存在资源问题，生产激活仍受下述 Linux 隔离、硬盘配额和 soak 门禁约束。 +本轮已覆盖 LangBot Core、Plugin Runtime 和 Box Runtime 的主要常驻对象、后台任务、队列、网络客户端、进程生命周期及数据库连接池。本轮定位到的攻击者可控或历史累积状态均已补充容量、超时、淘汰或确定性清理边界；修正后的高基数探针没有观察到随历史请求继续增长的活跃缓存。按本轮“代码审查 + 本地可重复测试”的验收口径，审查已经完成，未发现仍未处理的严重内存泄漏或 CPU 抢占路径。该结论不等于证明任意生产负载下不存在资源问题。 -代码级审查、跨仓全量测试和仓库 Dockerfile 构建的 Linux/cgroup v2 探针已经通过，但当前状态仍不能作为 Cloud 生产激活批准。完整剩余清单见 +最终 Cloud 拓扑和生产环境不是本轮完成条件。代码级审查、跨仓全量测试和仓库 Dockerfile 构建的 Linux/cgroup v2 探针已经通过，但当前状态仍不能单独作为 Cloud 生产激活批准。完整的环境侧剩余清单见 [Cloud v2 仍待验证事项](./cloud-v2-pending-verification.md)。其中与本轮资源审查直接相关、上线前还必须完成的项目包括： 1. 在最终 Cloud 部署权限和 cgroup 拓扑下重复 nsjail、namespace 和 delegated cgroup v2 的 CPU、内存、swap、PID、文件句柄验证。本轮一次性 Linux 容器已经证明代码路径可工作，但普通容器和仅 `--privileged` 的 private cgroup namespace 都不满足条件。 2. 为 Cloud Box 提供并验证硬文件系统 quota provider。普通 nsjail bind mount 不能证明总字节数和 inode 硬配额，当前严格 readiness 按设计会失败关闭。 3. 使用最终生产配置分布继续做容量测试，并据此确定单实例 Workspace placement 上限。本轮真实 PostgreSQL 16 + RLS 启动测试已经覆盖 1,000 个各带 Provider、三类 Model、Bot、Pipeline、KnowledgeBase、MCP 和 Plugin setting 的 Workspace，启动加载耗时和 SQL 次数保持线性；5,000 Workspace 的合成三代替换探针也证明旧运行时会释放。仓库已新增可同时采集 Core/Plugin/Box HTTP、进程树和 cgroup v2 的 24 小时门禁工具，并在受 CPU、memory、swap、PID 硬限制的 Linux 容器中完成短时自检；但最终生产候选拓扑的 24 小时运行仍未执行。测试中的 fake adapter/requester/Plugin handler 仍不能替代真实平台 SDK、外部连接池和插件进程的容量数据；合法活跃租户本身仍会线性占用内存。 -SDK 已先行发布到分支提交 `7c0b9827ed8597a1c84151b83fcf6307934fd944`，本提交集中的 LangBot +SDK 已先行发布到分支提交 `87d903e9b7c4e518ce066c89f65bbde4de825fc5`，本提交集中的 LangBot `pyproject.toml` 和 `uv.lock` 已精确钉住该提交。最终镜像仍需按待验证清单记录并核对实际安装版本。 ## 覆盖范围 @@ -124,6 +124,16 @@ SDK 已先行发布到分支提交 `7c0b9827ed8597a1c84151b83fcf6307934fd944`， - Monitoring 过期数据每表每轮默认最多删除 4 个批次、绝对最多 100 个批次；本地/S3 过期上传文件候选和每轮删除默认最多 1,000、绝对最多 10,000。单个历史数据量异常的 Workspace 不再能让一次维护循环无限物化候选或持续清空全部 backlog。 +- Workspace webhook 数量默认限制为 16、绝对限制为 64；管理查询和运行时 fan-out + 都只物化有界结果。实例同时发送的 webhook 请求默认限制为 16、绝对限制为 128； + 满载时直接跳过未获准目的地，不创建一批等待 semaphore 的 task。取消调用时会取消并 + await 已创建的所有请求任务，归还实例槽位。 +- Local/S3 Storage 的对象读取在实际 IO 中只读取 `limit + 1` 字节，S3 body 在成功、 + 超限和异常分支都会关闭；默认单对象 10 MiB、绝对上限 64 MiB。所有 scoped load + 以及 WebSocket attachment 都经过同一边界，写入也不能产生当前实例无法安全读取的对象。 +- Valkey Search 的批量删除改为固定页流式搜索、删除并累计计数，不再把全部匹配 key + 保留在 Python 列表；每次删除后从 offset 0 继续，避免结果集缩短造成跳项，并设置 + 1,000 轮绝对终止条件。 ### CPU 和事件循环保护 @@ -144,7 +154,7 @@ SDK 已先行发布到分支提交 `7c0b9827ed8597a1c84151b83fcf6307934fd944`， - S3 同步 SDK 使用线程执行，并通过实例级 semaphore 限制并发；默认 `storage.s3.max_concurrency=16`，可通过实例配置和环境变量覆写。 - Box 子进程 stderr 以 64 KiB 块读取，日志最多每秒输出 4 个摘录并汇总抑制数量，避免无换行或刷屏输出制造无界缓冲与日志放大。 - Plugin worker 日志单行最多保留 64 KiB；Box managed-process stdout relay 以固定 64 KiB 块读取，不再依赖换行符，避免超长无换行输出触发 `StreamReader` limit 或堵塞子进程。 -- Box generation fence 的代次更新改为只访问目标 Workspace 的 event 和 active-task 二级索引。原实现每次更新都会遍历全部 Workspace 的 fence/task 记录，10,000 个 Workspace 的第二阶段更新会退化为 O(N²) 并在 40 秒后仍未完成；修正后包括其他 SDK 高基数负载和本轮协议 offload 在内的当前完整双阶段探针耗时 `9.037s`。 +- Box generation fence 的代次更新改为只访问目标 Workspace 的 event 和 active-task 二级索引。原实现每次更新都会遍历全部 Workspace 的 fence/task 记录，10,000 个 Workspace 的第二阶段更新会退化为 O(N²) 并在 40 秒后仍未完成；修正后包括其他 SDK 高基数负载和本轮协议 offload 在内的当前完整双阶段探针耗时 `11.270s`。 - Box session 枚举、旧 generation 回收和 admission 计数均通过 Workspace 索引执行；admission 过期回收通过最小堆执行，不再在每次 RPC 上产生 O(实例总 session/grant 数) 的扫描。 - Model、Pipeline、RAG 和 Platform manager 均维护 Workspace 到运行时 key 的二级索引。Workspace generation 更新只清理目标 Workspace 的缓存和运行时，不再扫描实例内所有租户的 provider/model、pipeline、knowledge runtime 或 bot；回归测试使用禁止全局迭代的映射验证该边界。 - Cloud heartbeat 直接读取已加载且有容量边界的 Pipeline、MCP、KnowledgeBase 和 Bot registry 计数，不再为每个活跃 Workspace 依次打开 Tenant UoW、执行四类 COUNT 查询；这消除了租户数增长后每日周期性形成的串行 SQL/CPU 尖峰。OSS 模式仍保留数据库统计语义。 @@ -161,6 +171,13 @@ SDK 已先行发布到分支提交 `7c0b9827ed8597a1c84151b83fcf6307934fd944`， - Workspace 阻塞预算由服务端认证后的 `RequestContext`、公开 bot 的 RuntimeBot、公开对象 key 中经 binding fence 验证的 Workspace、Platform/TaskManager 的 ExecutionContext，以及 SDK 入站 ActionContext 建立，不接受调用方伪造的租户 header。公开 webhook、公开对象下载、Dashboard/Embed WebSocket、普通 HTTP handler、Platform adapter 和 detached tenant task 均已覆盖。容量拒绝在 Core HTTP 路径返回稳定的 429，health/debug counter 分开报告 global 与 scope rejection。 - Argon2 密码 hash/verify 只允许一个实例级在途操作，额外并发立即返回容量错误而不是在 asyncio semaphore 中无限积累等待请求；该 CPU/内存密集工作同时使用独立的 `system:authentication` 阻塞作用域。Cloud 本身仍禁用本地密码登录。 - WeCom 扩展 API 的无限客户端超时改为 120 秒；平台 webhook 的 AES、媒体 Base64 与同步 SDK 调用均移出共享事件循环。 +- 长文本转图片限制为 100,000 字符、256 行、800 万 RGBA 像素和 10 MiB 输出； + 超限时回退到 forward message。数字边界查找从重复 `count/find/sort` 改成线性扫描， + PIL image 使用显式关闭，压缩步长为零时也能终止。 +- Core 在每次 quota-enforced Box exec 前后遍历 Workspace 时使用非递归 DFS，并在 + 超过字节 quota 或默认 100,000/绝对 1,000,000 个目录项后立即停止；目录项洪泛 + 失败关闭，不再重复完整扫描 inode bomb。远程 outbox fallback 同时限制扫描项、 + 文件数、单文件和总字节，Python project manifest 使用分块 hash 并限制单文件 10 MiB。 ### 插件和 Box 资源隔离 @@ -170,6 +187,9 @@ SDK 已先行发布到分支提交 `7c0b9827ed8597a1c84151b83fcf6307934fd944`， - Box nsjail 的 cgroup v2 路径现在同时设置 `memory.max` 和 `memory.swap.max=0`。修复前，48 MiB 沙盒可以把强制提交的 128 MiB 页面换出并正常退出，形成宿主 swap 抢占；修复后同一探针以 exit 137 被 cgroup 杀死。 - 仓库 Docker Compose/Kubernetes 示例显式下发 Core、Plugin Runtime 和 Box Runtime 的 blocking executor 上限；Kubernetes Box readiness probe 从仅报告进程存活的 `/healthz` 改为 `/readyz`，使 backend 或 managed-mode 隔离检查失败时不会把 Pod 加入就绪流量。 - 相同 digest 的已验证代码和依赖环境可只读共享，每个 installation 的 home/tmp/data 和进程独立。 +- SDK 在发布共享依赖环境前最多校验 100,000 个目录项和 2 GiB 常规文件元数据总量； + 超限的 staging tree 会被原子清理而不会进入 worker。`requirements.txt` 和插件 + `manifest.yaml` 都使用 `limit + 1` 有界读取，manifest 额外限制为 1 MiB。 - Box session、managed process、completed process、admission record 和 RPC 文件均有实例级上限；Cloud entitlement 仍限制每个合资格 Workspace 一个 `global` session、零 managed process。 - Box Runtime 对上述实例级配置再增加不可放大的硬上限：session 5,000、managed process 1,024、completed process 10,000、admission record 250,000、RPC 单文件 @@ -206,6 +226,12 @@ SDK 已先行发布到分支提交 `7c0b9827ed8597a1c84151b83fcf6307934fd944`， - Core 与 SDK 各进程的通用阻塞 executor 默认使用 8 个 worker、128 个 pending 槽位、每 Workspace 4 个在途槽位；它是实例/进程级共享背压，不由 Workspace 或插件 manifest 调高，单 Workspace 配置硬性不得超过 worker 的一半。生产值应按容器 CPU 和上游阻塞时延校准，不能把 pending 当吞吐配置无限放大。 - 插件包下载上限 64 MiB，pip stdout/stderr 保留上限各 1 MiB；这不会限制安装进程实际输出，只限制父进程内存中的诊断副本。 - 通用远程响应和媒体默认上限 10 MiB；错误诊断正文只保留 4 KiB。Plugin binary storage 默认 10 MiB、绝对上限 64 MiB；Skill 文本、Plugin UI 和 host edit 分别限制为 1 MiB、4 MiB 和 1 MiB。 +- Storage scoped object 默认读写上限 10 MiB、代码绝对上限 64 MiB；Webhook 默认每 + Workspace 16 个、实例 16 个同时出站请求，代码绝对上限分别为 64 和 128。Box + Workspace quota 扫描默认最多访问 100,000 个目录项、绝对最多 1,000,000 个。 +- SDK 共享依赖环境在发布前最多接受 100,000 个条目、2 GiB 常规文件元数据总量； + artifact manifest 与 requirements 各最多 1 MiB。这些是 Runtime 控制面在启动 + worker 前的保护，不替代最终文件系统的 byte/inode 硬配额。 - Monitoring 查询上限由 `monitoring.query_limits` 配置并支持原生环境变量覆写，但始终 受代码绝对上限约束；cleanup 的每表批次数和 Storage 每轮文件数同样采用实例配置加 绝对上限。时间序列默认/绝对上限为 1,000/10,000 个数据库聚合桶，模型分组复用分页 @@ -220,8 +246,8 @@ SDK 已先行发布到分支提交 `7c0b9827ed8597a1c84151b83fcf6307934fd944`， | --- | --- | | LangBot Ruff + `git diff --check` | 通过 | | Plugin SDK Ruff + `git diff --check` | 通过 | -| LangBot 全量测试（使用远端精确钉住的新 SDK，含 unit/integration/Box/E2E） | `2839 passed, 33 skipped` | -| Plugin SDK 全量测试 | `1325 passed` | +| LangBot 全量测试（含 unit/integration/Box/E2E） | `2855 passed, 33 skipped` | +| Plugin SDK 全量测试 | `1328 passed` | | Space Go 全量测试与闭源 Cloud Adapter 测试 | Go `go test ./...` 通过；Adapter `40 passed` | | Space PostgreSQL 16 Cloud v2 目录与并发容量准入 | 通过；两个注册并发争用最后一个槽位时 `1 success / 1 capacity rejection / 1 active Workspace` | | Core PostgreSQL 16 Cloud runtime server timeout | 真实连接从 `pg_settings` 读回 `60000ms / 5000ms / 60000ms` 的 statement/lock/idle-transaction timeout，并显式 dispose | @@ -234,9 +260,9 @@ SDK 已先行发布到分支提交 `7c0b9827ed8597a1c84151b83fcf6307934fd944`， | Cloud soak 门禁解析/采样/判定单元测试 | `27 passed` | | Core/Plugin SDK event-loop monitor 专项测试 | 两仓各 `7 passed`，包含真实 50 ms scheduler stall | | Cloud soak Linux 硬限制短时自检 | 通过；CPU `0.5`、memory+swap `256 MiB`、PID `128` 均从 cgroup v2 读回，冷却尾段 verdict `pass` | -| Core 双阶段历史 churn 资源探针（使用精确钉住的新 SDK） | audit 通过，`11.895s` | -| Core 5,000 个 populated Workspace 三代容量探针（使用精确钉住的新 SDK） | 当前复跑通过，最大替换耗时比 `1.382` | -| Plugin SDK 双阶段资源探针 | audit 通过，`9.407s` | +| Core 双阶段历史 churn 资源探针（使用当前本地 SDK 分支） | audit 通过，`12.559s` | +| Core 5,000 个 populated Workspace 三代容量探针（使用当前本地 SDK 分支） | 当前复跑通过，最大替换耗时比 `1.405` | +| Plugin SDK 双阶段资源探针 | audit 通过，`11.270s` | 两个仓库新增了可重复执行的历史 churn 探针，Core 另有 populated Workspace 三代替换探针： @@ -266,14 +292,14 @@ Core audit 每个阶段执行 10,000 个空 Workspace 的真实 Model/Plugin man - 20,000 个限流身份：rate-limit container `10,000`。 - 10,000 个历史 task：task record `200`。 - 5,000 次 WebSocket churn：conversation 与 stream index 均为 `200`。 -- event-loop task、线程和文件描述符保持 `1 / 1 / 6`；使用远端精确钉住 SDK 的当前复跑中，第二阶段相对第一阶段 RSS 增长 `2,605,056 bytes`、tracemalloc current 增长 `344,622 bytes`，总耗时 `11.895s`。Session 淘汰改为 Workspace 索引和最小堆后，同一 audit 工作量相对此前 `16.150s` 明显下降。 +- event-loop task、线程和文件描述符保持 `1 / 1 / 6`；使用当前本地 SDK 分支的复跑中，第二阶段相对第一阶段 RSS 增长 `2,228,224 bytes`、tracemalloc current 增长 `344,669 bytes`，总耗时 `12.559s`。Session 淘汰改为 Workspace 索引和最小堆后，同一 audit 工作量相对此前 `16.150s` 明显下降。 Populated Workspace audit 为 5,000 个 Workspace 各加载一个 Provider、LLM、Embedding、Rerank、Pipeline、Bot、KnowledgeBase 和 MCP session，然后全部推进两个 generation： - 三个阶段的活跃 provider/model、pipeline、bot、knowledge 和 MCP registry 均精确维持 `5,000`，不存在按历史 generation 增长。 - 到第三阶段，前两代的 requester、Bot adapter 和 MCP session 各 `10,000` 个全部收到确定性关闭；weak reference 断言旧代对象可被回收。 - event-loop task、线程和文件描述符保持 `1 / 1 / 6`；使用远端精确钉住 SDK 的当前复跑中，第三阶段相对第二阶段 RSS 增长 `1,245,184 bytes`，tracemalloc current 仅增长 `2,061 bytes`。 -- 初始/第一次替换/第二次替换分别耗时 `1.660s / 2.151s / 2.296s`，最大替换耗时比为 `1.382`，未随历史代次出现 CPU 退化。 +- 初始/第一次替换/第二次替换分别耗时 `1.893s / 2.549s / 2.659s`，最大替换耗时比为 `1.405`，未随历史代次出现 CPU 退化。 - macOS RSS sample 从初始的 `154,648,576` 增至第一阶段 `368,181,248`、第二阶段 `389,087,232` 和第三阶段 `390,332,416 bytes`；第二次替换只比第一次替换增加约 1.19 MiB，但“合法活跃租户资源的线性容量”仍必须作为 placement 容量输入。这里使用轻量 fake adapter/requester，不应把第一阶段约 204 MiB 增量外推为生产每租户成本。 Plugin SDK audit 每个阶段执行 25,000 次 loopback RPC、5,000 次安装 binding 激活/撤销、10,000 个 Workspace generation 更新和 2,500 次带 Workspace 上下文的 Box session 创建/删除。第一、第二阶段的保留状态完全一致： @@ -281,7 +307,7 @@ Plugin SDK audit 每个阶段执行 25,000 次 loopback RPC、5,000 次安装 bi - RPC waiter、stream queue、action task 和活跃 installation binding 均为 `0`。 - installation watermark 为有界的 `5,000`；Workspace generation record 为有界的 `10,000`，没有等待者时 generation event 为 `0`。 - generation active task/index、Box session、Box Workspace session index、creating/closing/background task 和 session lock 均为 `0`。 -- event-loop task 和文件描述符保持 `1 / 7`；当前复跑第二阶段相对第一阶段 RSS peak 增长 `2,670,592 bytes`、tracemalloc current 增长 `289,746 bytes`，总耗时 `9.407s`。耗时增加来自本轮把大协议消息的 JSON/Pydantic、UTF-8 编码、分片和拼接移入有界线程池；25,000 RPC/阶段的合成探针仍约为 5,300 RPC/s，结构状态和第二阶段 tracemalloc 增量保持平稳。 +- event-loop task 和文件描述符保持 `1 / 7`；当前复跑第二阶段相对第一阶段 RSS peak 增长 `2,637,824 bytes`、tracemalloc current 增长 `289,746 bytes`，总耗时 `11.270s`。耗时增加来自本轮把大协议消息的 JSON/Pydantic、UTF-8 编码、分片和拼接移入有界线程池；结构状态和第二阶段 tracemalloc 增量保持平稳。 第二轮反向静态审查另外枚举了 Core 的 50 个显式 task 创建点和 204 个线程、阻塞调用及子进程调用点，以及 SDK 的 28 个显式 task 创建点和 62 个线程、阻塞调用及子进程调用点。第三轮独立复核继续从高基数定时器、目录遍历、准入全表扫描和取消竞态反推，新增关闭了 Plugin restart 冷却唤醒群、MCP idle 数据库轮询、nsjail orphan 的 O(session × process) 启动扫描、message aggregation 的 O(buffer) 准入及 Skill inode/文本列表边界。显式 task 均具有持有者、完成回调或 `finally` 回收路径；所有生产入口在第一次 `asyncio.to_thread()` 前安装有界默认 executor。Core、Plugin Runtime 和 Box 的公开 `/healthz`（Box `/readyz` 亦同）会输出各自的 aggregate runtime/resource counter 和 event-loop lag，供 soak 对比活跃量、pending、累计 capacity rejection 与调度延迟；不输出 debug key、控制 token、租户或插件身份。Plugin Runtime 的授权 debug info 复用同一资源快照，避免公开/私有指标语义漂移。 diff --git a/pyproject.toml b/pyproject.toml index 2864c76c5..fee880288 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,7 +71,7 @@ dependencies = [ "chromadb>=1.0.0,<2.0.0", "qdrant-client (>=1.15.1,<2.0.0)", "pyseekdb==1.1.0.post3", - "langbot-plugin @ git+https://github.com/langbot-app/langbot-plugin-sdk.git@7c0b9827ed8597a1c84151b83fcf6307934fd944", + "langbot-plugin @ git+https://github.com/langbot-app/langbot-plugin-sdk.git@87d903e9b7c4e518ce066c89f65bbde4de825fc5", "asyncpg>=0.30.0", "line-bot-sdk>=3.19.0", "matrix-nio>=0.25.2", diff --git a/src/langbot/pkg/api/http/service/webhook.py b/src/langbot/pkg/api/http/service/webhook.py index 14b029864..1925b51cd 100644 --- a/src/langbot/pkg/api/http/service/webhook.py +++ b/src/langbot/pkg/api/http/service/webhook.py @@ -8,12 +8,31 @@ from .secrets import SECRET_MASK, mask_secret_value, restore_secret_placeholders from .tenant import TenantContext, require_workspace_uuid, scope_statement +_DEFAULT_MAX_WEBHOOKS_PER_WORKSPACE = 16 +_HARD_MAX_WEBHOOKS_PER_WORKSPACE = 64 + + class WebhookService: ap: app.Application def __init__(self, ap: app.Application) -> None: self.ap = ap + def max_per_workspace(self) -> int: + """Return the configured webhook cap within the process hard limit.""" + + config = getattr(getattr(self.ap, 'instance_config', None), 'data', {}) + try: + value = int( + config.get('webhooks', {}).get( + 'max_per_workspace', + _DEFAULT_MAX_WEBHOOKS_PER_WORKSPACE, + ) + ) + except (AttributeError, TypeError, ValueError): + value = _DEFAULT_MAX_WEBHOOKS_PER_WORKSPACE + return min(max(value, 1), _HARD_MAX_WEBHOOKS_PER_WORKSPACE) + def _serialize_webhook(self, entity, *, include_secret: bool) -> dict: serialized = self.ap.persistence_mgr.serialize_model(webhook.Webhook, entity) if not include_secret: @@ -24,7 +43,11 @@ class WebhookService: async def get_webhooks(self, context: TenantContext, *, include_secret: bool = False) -> list[dict]: """Get all webhooks""" result = await self.ap.persistence_mgr.execute_async( - scope_statement(sqlalchemy.select(webhook.Webhook), webhook.Webhook, context) + scope_statement( + sqlalchemy.select(webhook.Webhook).order_by(webhook.Webhook.id).limit(_HARD_MAX_WEBHOOKS_PER_WORKSPACE), + webhook.Webhook, + context, + ) ) webhooks = result.all() @@ -40,6 +63,15 @@ class WebhookService: ) -> dict: """Create a new webhook""" workspace_uuid = require_workspace_uuid(context) + max_webhooks = self.max_per_workspace() + count_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(sqlalchemy.func.count()) + .select_from(webhook.Webhook) + .where(webhook.Webhook.workspace_uuid == workspace_uuid) + ) + if (count_result.scalar() or 0) >= max_webhooks: + raise ValueError(f'Maximum number of webhooks ({max_webhooks}) reached') + url = restore_secret_placeholders(url, sensitive=True) webhook_data = { 'workspace_uuid': workspace_uuid, @@ -143,6 +175,8 @@ class WebhookService: webhook.Webhook, context, ) + .order_by(webhook.Webhook.id) + .limit(self.max_per_workspace()) ) webhooks = result.all() diff --git a/src/langbot/pkg/box/service.py b/src/langbot/pkg/box/service.py index 65022482e..56c8ae59b 100644 --- a/src/langbot/pkg/box/service.py +++ b/src/langbot/pkg/box/service.py @@ -38,6 +38,8 @@ _INT_ADAPTER = pydantic.TypeAdapter(int) _UTC = _dt.timezone.utc _MAX_RECENT_ERRORS = 50 _MIB = 1024 * 1024 +_DEFAULT_MAX_WORKSPACE_ENTRIES = 100_000 +_HARD_MAX_WORKSPACE_ENTRIES = 1_000_000 def _create_shared_workspace_probe(root: str, marker_name: str, payload: bytes) -> None: @@ -1214,25 +1216,53 @@ class BoxService: import json as _json target_dir = f'{self.OUTBOX_MOUNT_DIR}/{self._attachment_query_key(query)}' - max_bytes = self._EXEC_FALLBACK_MAX_BYTES + max_file_bytes = self._EXEC_FALLBACK_MAX_BYTES + max_files = self._ATTACHMENT_MAX_FILES + max_total_bytes = max_file_bytes * max_files + max_scan_entries = 1000 script = ( 'import base64, json, os\n' f'target = {target_dir!r}\n' - f'max_bytes = {max_bytes}\n' + f'max_file_bytes = {max_file_bytes}\n' + f'max_files = {max_files}\n' + f'max_total_bytes = {max_total_bytes}\n' + f'max_scan_entries = {max_scan_entries}\n' 'out = []\n' + 'total_bytes = 0\n' + 'scanned_entries = 0\n' + 'stack = [target]\n' 'if os.path.isdir(target):\n' - ' for root, _dirs, names in os.walk(target):\n' - ' for n in sorted(names):\n' - ' p = os.path.join(root, n)\n' + ' while stack and len(out) < max_files and scanned_entries < max_scan_entries:\n' + ' current = stack.pop()\n' + ' try:\n' + ' with os.scandir(current) as iterator:\n' + ' entries = sorted(iterator, key=lambda item: item.name, reverse=True)\n' + ' except OSError:\n' + ' continue\n' + ' for entry in entries:\n' + ' scanned_entries += 1\n' + ' if scanned_entries > max_scan_entries:\n' + ' break\n' ' try:\n' - ' if os.path.getsize(p) > max_bytes:\n' + ' if entry.is_dir(follow_symlinks=False):\n' + ' stack.append(entry.path)\n' + ' continue\n' + ' if not entry.is_file(follow_symlinks=False):\n' + ' continue\n' + ' size = entry.stat(follow_symlinks=False).st_size\n' + ' if size > max_file_bytes or total_bytes + size > max_total_bytes:\n' + ' continue\n' + " with open(entry.path, 'rb') as f:\n" + ' data = f.read(max_file_bytes + 1)\n' + ' if len(data) > max_file_bytes or total_bytes + len(data) > max_total_bytes:\n' ' continue\n' - " with open(p, 'rb') as f:\n" - ' data = f.read()\n' ' except OSError:\n' ' continue\n' - ' rel = os.path.relpath(p, target)\n' + ' rel = os.path.relpath(entry.path, target)\n' " out.append({'name': rel, 'b64': base64.b64encode(data).decode('ascii')})\n" + ' total_bytes += len(data)\n' + ' if len(out) >= max_files:\n' + ' break\n' 'print(json.dumps(out))\n' ) result = await self.execute_tool( @@ -1888,29 +1918,50 @@ class BoxService: if normalized_timeout > profile.max_timeout_sec: params['timeout_sec'] = profile.max_timeout_sec - def _get_workspace_size_bytes(self, root: str) -> int: - total = 0 + def _max_workspace_entries(self) -> int: + data = getattr(getattr(self.ap, 'instance_config', None), 'data', {}) + try: + configured = int( + data.get('box', {}).get('limits', {}).get('max_workspace_entries', _DEFAULT_MAX_WORKSPACE_ENTRIES) + ) + except (AttributeError, TypeError, ValueError): + configured = _DEFAULT_MAX_WORKSPACE_ENTRIES + return min(max(configured, 1), _HARD_MAX_WORKSPACE_ENTRIES) - def _walk(path: str): - nonlocal total + @staticmethod + def _get_workspace_usage( + root: str, + *, + stop_after_bytes: int, + max_entries: int, + ) -> tuple[int, int, bool]: + """Scan depth-first without recursion and stop at either hard bound.""" + + total = 0 + entries_seen = 0 + directories = [root] + while directories: + path = directories.pop() try: with os.scandir(path) as entries: for entry in entries: + entries_seen += 1 + if entries_seen > max_entries: + return total, entries_seen, True try: if entry.is_symlink(): total += entry.stat(follow_symlinks=False).st_size - continue - if entry.is_dir(follow_symlinks=False): - _walk(entry.path) - continue - total += entry.stat(follow_symlinks=False).st_size + elif entry.is_dir(follow_symlinks=False): + directories.append(entry.path) + else: + total += entry.stat(follow_symlinks=False).st_size except FileNotFoundError: continue + if total > stop_after_bytes: + return total, entries_seen, False except FileNotFoundError: - return - - _walk(root) - return total + continue + return total, entries_seen, False async def _enforce_workspace_quota(self, spec: BoxSpec, *, phase: str) -> None: if spec.host_path is None or spec.workspace_quota_mb <= 0: @@ -1923,15 +1974,26 @@ class BoxService: # Walk the workspace off the event loop — this runs on every # quota-enforced exec, and a large tree would otherwise block the whole # asyncio runtime (all bots/pipelines) for the duration of the scan. - used_bytes = await asyncio.to_thread(self._get_workspace_size_bytes, host_path) limit_bytes = spec.workspace_quota_mb * _MIB + max_entries = self._max_workspace_entries() + used_bytes, entries_seen, entry_limit_exceeded = await asyncio.to_thread( + self._get_workspace_usage, + host_path, + stop_after_bytes=limit_bytes, + max_entries=max_entries, + ) + if entry_limit_exceeded: + raise BoxValidationError( + f'workspace entry limit exceeded {phase}: ' + f'entries>{max_entries} host_path={host_path} session_id={spec.session_id}' + ) if used_bytes <= limit_bytes: return raise BoxValidationError( f'workspace quota exceeded {phase}: ' f'used={used_bytes} bytes limit={limit_bytes} bytes ' - f'host_path={host_path} session_id={spec.session_id}' + f'entries={entries_seen} host_path={host_path} session_id={spec.session_id}' ) async def _cleanup_exceeded_session(self, context: TenantContext, spec: BoxSpec) -> None: diff --git a/src/langbot/pkg/box/workspace.py b/src/langbot/pkg/box/workspace.py index f167f7455..3dd91c67b 100644 --- a/src/langbot/pkg/box/workspace.py +++ b/src/langbot/pkg/box/workspace.py @@ -171,17 +171,23 @@ def wrap_python_command_with_env( import sys root = "{mount_path}" + max_manifest_bytes = 10 * 1024 * 1024 digest = hashlib.sha256() manifest_files = [] for rel in ("requirements.txt", "pyproject.toml", "setup.py", "setup.cfg"): path = os.path.join(root, rel) if not os.path.isfile(path): continue + if os.path.getsize(path) > max_manifest_bytes: + raise RuntimeError( + f"Python project manifest exceeds {{max_manifest_bytes}} bytes: {{rel}}" + ) manifest_files.append(rel) with open(path, "rb") as handle: digest.update(rel.encode("utf-8")) digest.update(b"\\0") - digest.update(handle.read()) + while chunk := handle.read(1024 * 1024): + digest.update(chunk) digest.update(b"\\0") print( diff --git a/src/langbot/pkg/core/stages/load_config.py b/src/langbot/pkg/core/stages/load_config.py index a0227e905..a89cb0b48 100644 --- a/src/langbot/pkg/core/stages/load_config.py +++ b/src/langbot/pkg/core/stages/load_config.py @@ -69,7 +69,19 @@ _RUNTIME_POLICY_DEFAULTS = { }, 'auto_cleanup': {'max_batches_per_table_per_run': 4}, }, - 'storage': {'cleanup': {'max_files_per_run': 1000}}, + 'storage': { + 'max_object_read_bytes': 10485760, + 'cleanup': {'max_files_per_run': 1000}, + }, + 'webhooks': { + 'max_per_workspace': 16, + 'max_inflight_requests': 16, + }, + 'box': { + 'limits': { + 'max_workspace_entries': 100000, + } + }, } diff --git a/src/langbot/pkg/pipeline/longtext/strategies/image.py b/src/langbot/pkg/pipeline/longtext/strategies/image.py index 10be8998c..8b25ee455 100644 --- a/src/langbot/pkg/pipeline/longtext/strategies/image.py +++ b/src/langbot/pkg/pipeline/longtext/strategies/image.py @@ -12,10 +12,21 @@ from PIL import Image, ImageDraw, ImageFont import functools from .. import strategy as strategy_model +from .forward import ForwardComponentStrategy import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query import langbot_plugin.api.entities.builtin.platform.message as platform_message +_MAX_TEXT_TO_IMAGE_CHARS = 100000 +_MAX_TEXT_TO_IMAGE_LINES = 256 +_MAX_TEXT_TO_IMAGE_PIXELS = 8_000_000 +_MAX_RENDERED_IMAGE_BYTES = 10 * 1024 * 1024 + + +class _TextToImageCapacityError(ValueError): + """The requested image would exceed a deterministic resource boundary.""" + + @strategy_model.strategy_class('image') class Text2ImageStrategy(strategy_model.LongTextStrategy): async def initialize(self): @@ -30,6 +41,12 @@ class Text2ImageStrategy(strategy_model.LongTextStrategy): ) async def process(self, message: str, query: pipeline_query.Query) -> list[platform_message.MessageComponent]: + if len(message) > _MAX_TEXT_TO_IMAGE_CHARS: + self.ap.logger.warning( + f'Text-to-image input exceeds {_MAX_TEXT_TO_IMAGE_CHARS} characters; using forward message' + ) + return await ForwardComponentStrategy(self.ap).process(message, query) + def render() -> str: render_id = f'{int(time.time())}-{uuid.uuid4().hex}' img_path = f'temp/{render_id}.png' @@ -45,7 +62,12 @@ class Text2ImageStrategy(strategy_model.LongTextStrategy): outfile=compressed_path, ) with open(compressed_path, 'rb') as f: - return base64.b64encode(f.read()).decode('utf-8') + image_bytes = f.read(_MAX_RENDERED_IMAGE_BYTES + 1) + if len(image_bytes) > _MAX_RENDERED_IMAGE_BYTES: + raise _TextToImageCapacityError( + f'Rendered image exceeds the {_MAX_RENDERED_IMAGE_BYTES}-byte limit' + ) + return base64.b64encode(image_bytes).decode('utf-8') finally: for path in {img_path, compressed_path}: if os.path.exists(path): @@ -53,7 +75,11 @@ class Text2ImageStrategy(strategy_model.LongTextStrategy): # Font measurement, image rendering and compression are CPU-bound PIL # work and must not block the shared asyncio loop for every tenant. - image_base64 = await asyncio.to_thread(render) + try: + image_base64 = await asyncio.to_thread(render) + except _TextToImageCapacityError as exc: + self.ap.logger.warning(f'{exc}; using forward message') + return await ForwardComponentStrategy(self.ap).process(message, query) return [ platform_message.Image( @@ -67,38 +93,7 @@ class Text2ImageStrategy(strategy_model.LongTextStrategy): :param path:目标字符串 :return:: : [['1', 16], ['2', 35], ['1', 51]] """ - kv = [] - nums = [] - beforeDatas = re.findall('[\\d]+', path) - for num in beforeDatas: - indexV = [] - times = path.count(num) - if times > 1: - if num not in nums: - indexs = re.finditer(num, path) - for index in indexs: - iV = [] - i = index.span()[0] - iV.append(num) - iV.append(i) - kv.append(iV) - nums.append(num) - else: - index = path.find(num) - indexV.append(num) - indexV.append(index) - kv.append(indexV) - # 根据数字位置排序 - indexSort = [] - resultIndex = [] - for vi in kv: - indexSort.append(vi[1]) - indexSort.sort() - for i in indexSort: - for v in kv: - if i == v[1]: - resultIndex.append(v) - return resultIndex + return [[match.group(0), match.start()] for match in re.finditer(r'\d+', path)] def get_size(self, file): # 获取文件大小:KB @@ -126,9 +121,9 @@ class Text2ImageStrategy(strategy_model.LongTextStrategy): return infile, o_size outfile = self.get_outfile(infile, outfile) while o_size > kb: - im = Image.open(infile) - im.save(outfile, quality=quality) - if quality - step < 0: + with Image.open(infile) as im: + im.save(outfile, quality=quality) + if step <= 0 or quality - step < 0: break quality -= step o_size = self.get_size(outfile) @@ -137,12 +132,21 @@ class Text2ImageStrategy(strategy_model.LongTextStrategy): def _split_text_lines(self, text_str: str, text_width: int, font) -> list[str]: """Split text while guaranteeing that every loop iteration advances.""" + if len(text_str) > _MAX_TEXT_TO_IMAGE_CHARS: + raise _TextToImageCapacityError(f'Text-to-image input exceeds {_MAX_TEXT_TO_IMAGE_CHARS} characters') + final_lines: list[str] = [] + + def append_line(value: str) -> None: + if len(final_lines) >= _MAX_TEXT_TO_IMAGE_LINES: + raise _TextToImageCapacityError(f'Text-to-image output exceeds {_MAX_TEXT_TO_IMAGE_LINES} lines') + final_lines.append(value) + text_width = max(int(text_width), 1) for line in text_str.replace('\t', ' ').split('\n'): line_width = font.getlength(line) if not line or line_width < text_width: - final_lines.append(line) + append_line(line) continue rest_text = line @@ -151,16 +155,18 @@ class Text2ImageStrategy(strategy_model.LongTextStrategy): point = int(len(rest_text) * (text_width / line_width)) point = max(1, min(point, len(rest_text))) - for number, number_index in self.indexNumber(rest_text): - if number_index < point < number_index + len(number) and number_index != 0: - point = number_index - break + if 0 < point < len(rest_text) and rest_text[point - 1].isdigit() and rest_text[point].isdigit(): + number_start = point - 1 + while number_start > 0 and rest_text[number_start - 1].isdigit(): + number_start -= 1 + if number_start > 0: + point = number_start point = max(1, min(point, len(rest_text))) - final_lines.append(rest_text[:point]) + append_line(rest_text[:point]) rest_text = rest_text[point:] if rest_text and font.getlength(rest_text) < text_width: - final_lines.append(rest_text) + append_line(rest_text) break return final_lines @@ -171,38 +177,34 @@ class Text2ImageStrategy(strategy_model.LongTextStrategy): width=800, query: pipeline_query.Query = None, ): + width = int(width) + if width < 1: + raise _TextToImageCapacityError('Text-to-image width must be positive') font = self.get_font(query.pipeline_config['output']['long-text-processing']['font-path']) text_width = max(width - 80, 1) final_lines = self._split_text_lines(text_str, text_width, font) + image_height = max(280, len(final_lines) * 35 + 65) + if width * image_height > _MAX_TEXT_TO_IMAGE_PIXELS: + raise _TextToImageCapacityError(f'Text-to-image canvas exceeds the {_MAX_TEXT_TO_IMAGE_PIXELS}-pixel limit') # 准备画布 - img = Image.new('RGBA', (width, max(280, len(final_lines) * 35 + 65)), (255, 255, 255, 255)) - draw = ImageDraw.Draw(img, mode='RGBA') + img = Image.new('RGBA', (width, image_height), (255, 255, 255, 255)) + try: + draw = ImageDraw.Draw(img, mode='RGBA') - self.ap.logger.debug('正在绘制图片...') - # 绘制正文 - line_number = 0 - offset_x = 20 - offset_y = 30 - for final_line in final_lines: - draw.text( - (offset_x, offset_y + 35 * line_number), - final_line, - fill=(0, 0, 0), - font=font, - ) - # 遍历此行,检查是否有emoji - idx_in_line = 0 - for ch in final_line: - # 检查字符占位宽 - char_code = ord(ch) - if char_code >= 127: - idx_in_line += 1 - else: - idx_in_line += 0.5 + self.ap.logger.debug('正在绘制图片...') + offset_x = 20 + offset_y = 30 + for line_number, final_line in enumerate(final_lines): + draw.text( + (offset_x, offset_y + 35 * line_number), + final_line, + fill=(0, 0, 0), + font=font, + ) - line_number += 1 - - self.ap.logger.debug('正在保存图片...') - img.save(save_as) + self.ap.logger.debug('正在保存图片...') + img.save(save_as) + finally: + img.close() return save_as diff --git a/src/langbot/pkg/platform/sources/websocket_adapter.py b/src/langbot/pkg/platform/sources/websocket_adapter.py index dcda32120..6176e2d93 100644 --- a/src/langbot/pkg/platform/sources/websocket_adapter.py +++ b/src/langbot/pkg/platform/sources/websocket_adapter.py @@ -564,7 +564,11 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter) raise ValueError('Attachment key does not belong to this WebSocket connection') try: - file_content = await storage_mgr.storage_provider.load(comp_path) + file_content = await storage_mgr.load_scoped_object_key( + execution_context, + comp_path, + expected_owner_type='upload_image', + ) base64_str = (await asyncio.to_thread(base64.b64encode, file_content)).decode('utf-8') lowered = comp_path.lower() diff --git a/src/langbot/pkg/platform/webhook_pusher.py b/src/langbot/pkg/platform/webhook_pusher.py index a6af2284a..83f5cee11 100644 --- a/src/langbot/pkg/platform/webhook_pusher.py +++ b/src/langbot/pkg/platform/webhook_pusher.py @@ -15,6 +15,10 @@ if TYPE_CHECKING: import langbot_plugin.api.entities.builtin.platform.events as platform_events +_DEFAULT_MAX_INFLIGHT_WEBHOOK_REQUESTS = 16 +_HARD_MAX_INFLIGHT_WEBHOOK_REQUESTS = 128 + + class WebhookPusher: """Push bot events to configured webhooks""" @@ -24,6 +28,56 @@ class WebhookPusher: def __init__(self, ap: app.Application): self.ap = ap self.logger = self.ap.logger + self._delivery_lock = asyncio.Lock() + self._inflight_requests = 0 + + def _max_inflight_requests(self) -> int: + config = getattr(getattr(self.ap, 'instance_config', None), 'data', {}) + try: + value = int( + config.get('webhooks', {}).get( + 'max_inflight_requests', + _DEFAULT_MAX_INFLIGHT_WEBHOOK_REQUESTS, + ) + ) + except (AttributeError, TypeError, ValueError): + value = _DEFAULT_MAX_INFLIGHT_WEBHOOK_REQUESTS + return min(max(value, 1), _HARD_MAX_INFLIGHT_WEBHOOK_REQUESTS) + + async def _reserve_delivery_slots(self, requested: int) -> int: + async with self._delivery_lock: + available = max(self._max_inflight_requests() - self._inflight_requests, 0) + admitted = min(max(requested, 0), available) + self._inflight_requests += admitted + return admitted + + async def _release_delivery_slots(self, released: int) -> None: + async with self._delivery_lock: + self._inflight_requests = max(self._inflight_requests - released, 0) + + async def _push_to_webhooks(self, webhooks: list[dict], payload: dict) -> list[object]: + """Dispatch only requests admitted by the instance-wide hard bound.""" + + admitted = await self._reserve_delivery_slots(len(webhooks)) + if admitted < len(webhooks): + self.logger.warning( + 'Webhook delivery capacity reached; skipped %d of %d destinations', + len(webhooks) - admitted, + len(webhooks), + ) + if admitted == 0: + return [] + + tasks = [asyncio.create_task(self._push_to_webhook(webhook['url'], payload)) for webhook in webhooks[:admitted]] + try: + return await asyncio.gather(*tasks, return_exceptions=True) + except asyncio.CancelledError: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise + finally: + await self._release_delivery_slots(admitted) async def push_person_message( self, @@ -58,9 +112,7 @@ class WebhookPusher: }, } - # Push to all webhooks asynchronously - tasks = [self._push_to_webhook(webhook['url'], payload) for webhook in webhooks] - results = await asyncio.gather(*tasks, return_exceptions=True) + results = await self._push_to_webhooks(webhooks, payload) # Check if any webhook responded with skip_pipeline=true for result in results: @@ -111,9 +163,7 @@ class WebhookPusher: }, } - # Push to all webhooks asynchronously - tasks = [self._push_to_webhook(webhook['url'], payload) for webhook in webhooks] - results = await asyncio.gather(*tasks, return_exceptions=True) + results = await self._push_to_webhooks(webhooks, payload) # Check if any webhook responded with skip_pipeline=true for result in results: diff --git a/src/langbot/pkg/provider/tools/loaders/native.py b/src/langbot/pkg/provider/tools/loaders/native.py index 3b4e68cea..d2dfdc969 100644 --- a/src/langbot/pkg/provider/tools/loaders/native.py +++ b/src/langbot/pkg/provider/tools/loaders/native.py @@ -905,17 +905,27 @@ if not path.startswith('/workspace'): print(json.dumps({{'ok': False, 'error': 'Path must be under /workspace.'}})) elif not os.path.isfile(path): print(json.dumps({{'ok': False, 'error': f'File not found: {{path}}'}})) +elif os.path.getsize(path) > {_MAX_HOST_EDIT_FILE_BYTES}: + print(json.dumps({{'ok': False, 'error': 'File exceeds the {_MAX_HOST_EDIT_FILE_BYTES}-byte edit limit.'}})) else: - with open(path, 'r', encoding='utf-8', errors='replace') as f: - content = f.read() + with open(path, 'rb') as f: + raw_content = f.read({_MAX_HOST_EDIT_FILE_BYTES + 1}) + if len(raw_content) > {_MAX_HOST_EDIT_FILE_BYTES}: + print(json.dumps({{'ok': False, 'error': 'File exceeds the {_MAX_HOST_EDIT_FILE_BYTES}-byte edit limit.'}})) + raise SystemExit(0) + content = raw_content.decode('utf-8', errors='replace') count = content.count(old_string) if count == 0: print(json.dumps({{'ok': False, 'error': 'old_string not found in file.'}})) elif count > 1: print(json.dumps({{'ok': False, 'error': f'old_string matches {{count}} locations; provide a more unique string.'}})) else: + new_content = content.replace(old_string, new_string, 1) + if len(new_content.encode('utf-8')) > {_MAX_HOST_EDIT_FILE_BYTES}: + print(json.dumps({{'ok': False, 'error': 'Edited file exceeds the {_MAX_HOST_EDIT_FILE_BYTES}-byte limit.'}})) + raise SystemExit(0) with open(path, 'w', encoding='utf-8') as f: - f.write(content.replace(old_string, new_string, 1)) + f.write(new_content) print(json.dumps({{'ok': True, 'path': path}})) """.strip() return await self._run_workspace_file_script(script, query) diff --git a/src/langbot/pkg/storage/mgr.py b/src/langbot/pkg/storage/mgr.py index 1ec20375c..c6c7c8a93 100644 --- a/src/langbot/pkg/storage/mgr.py +++ b/src/langbot/pkg/storage/mgr.py @@ -14,6 +14,7 @@ from .providers import localstorage _SAFE_OWNER_TYPE = re.compile(r'^[a-z][a-z0-9_-]{0,63}$') +_DEFAULT_OBJECT_READ_BYTES = 10 * 1024 * 1024 _SCOPED_KEY = re.compile( r'^v1/(?P[a-f0-9]{24})/' r'(?P[0-9a-fA-F-]{36})/' @@ -34,6 +35,35 @@ class StorageMgr: def __init__(self, ap: app.Application): self.ap = ap + def _object_read_limit(self) -> int: + config = getattr(getattr(self.ap, 'instance_config', None), 'data', {}) + try: + configured = int( + config.get('storage', {}).get( + 'max_object_read_bytes', + _DEFAULT_OBJECT_READ_BYTES, + ) + ) + except (AttributeError, TypeError, ValueError): + configured = _DEFAULT_OBJECT_READ_BYTES + return min(max(configured, 1), provider.HARD_MAX_STORAGE_OBJECT_BYTES) + + async def _load_object_bounded(self, object_key: str) -> bytes: + max_bytes = self._object_read_limit() + bounded_loader = getattr(self.storage_provider, 'load_bounded', None) + if callable(bounded_loader): + return await bounded_loader(object_key, max_bytes=max_bytes) + + # Compatibility for lightweight and third-party providers. Built-in + # providers enforce the same bound in the actual read operation. + object_size = await self.storage_provider.size(object_key) + if object_size > max_bytes: + raise ValueError(f'Storage object exceeds the {max_bytes}-byte read limit') + value = await self.storage_provider.load(object_key) + if len(value) > max_bytes: + raise ValueError(f'Storage object exceeds the {max_bytes}-byte read limit') + return value + @staticmethod def _require_execution_scope( context: ExecutionContext | RequestContext, @@ -150,6 +180,9 @@ class StorageMgr: preserve_suffix: bool = True, ) -> str: await self._require_active_execution_scope(context) + max_bytes = self._object_read_limit() + if len(value) > max_bytes: + raise ValueError(f'Storage object exceeds the {max_bytes}-byte write limit') object_key = self.scoped_object_key( context, owner_type=owner_type, @@ -177,7 +210,7 @@ class StorageMgr: key=key, preserve_suffix=preserve_suffix, ) - return await self.storage_provider.load(object_key) + return await self._load_object_bounded(object_key) async def delete_scoped( self, @@ -223,7 +256,7 @@ class StorageMgr: return None if not await self.storage_provider.exists(object_key): return None - return await self.storage_provider.load(object_key) + return await self._load_object_bounded(object_key) @classmethod def require_scoped_object_key( @@ -274,7 +307,7 @@ class StorageMgr: object_key, expected_owner_type=expected_owner_type, ) - return await self.storage_provider.load(object_key) + return await self._load_object_bounded(object_key) async def size_scoped_object_key( self, diff --git a/src/langbot/pkg/storage/provider.py b/src/langbot/pkg/storage/provider.py index 4e46f09c7..1747d49d2 100644 --- a/src/langbot/pkg/storage/provider.py +++ b/src/langbot/pkg/storage/provider.py @@ -5,6 +5,19 @@ import abc from ..core import app +HARD_MAX_STORAGE_OBJECT_BYTES = 64 * 1024 * 1024 + + +def normalize_read_limit(max_bytes: int) -> int: + """Validate a provider read limit without allowing callers to bypass the hard cap.""" + + try: + normalized = int(max_bytes) + except (TypeError, ValueError): + normalized = HARD_MAX_STORAGE_OBJECT_BYTES + return min(max(normalized, 1), HARD_MAX_STORAGE_OBJECT_BYTES) + + class StorageProvider(abc.ABC): ap: app.Application @@ -34,6 +47,23 @@ class StorageProvider(abc.ABC): ) -> bytes: pass + async def load_bounded(self, key: str, *, max_bytes: int) -> bytes: + """Fallback for third-party providers that have not implemented streaming bounds. + + Built-in providers override this method so the byte limit is enforced by + the actual read. The size check still protects compatible providers from + downloading a known oversized object. + """ + + max_bytes = normalize_read_limit(max_bytes) + object_size = await self.size(key) + if object_size > max_bytes: + raise ValueError(f'Storage object exceeds the {max_bytes}-byte read limit') + value = await self.load(key) + if len(value) > max_bytes: + raise ValueError(f'Storage object exceeds the {max_bytes}-byte read limit') + return value + @abc.abstractmethod async def exists( self, diff --git a/src/langbot/pkg/storage/providers/localstorage.py b/src/langbot/pkg/storage/providers/localstorage.py index c64698232..9775badb5 100644 --- a/src/langbot/pkg/storage/providers/localstorage.py +++ b/src/langbot/pkg/storage/providers/localstorage.py @@ -51,9 +51,21 @@ class LocalStorageProvider(provider.StorageProvider): self, key: str, ) -> bytes: + return await self.load_bounded(key, max_bytes=provider.HARD_MAX_STORAGE_OBJECT_BYTES) + + async def load_bounded( + self, + key: str, + *, + max_bytes: int, + ) -> bytes: + max_bytes = provider.normalize_read_limit(max_bytes) resolved = await asyncio.to_thread(_safe_resolve, LOCAL_STORAGE_PATH, key) async with aiofiles.open(resolved, 'rb') as f: - return await f.read() + value = await f.read(max_bytes + 1) + if len(value) > max_bytes: + raise ValueError(f'Storage object exceeds the {max_bytes}-byte read limit') + return value async def exists( self, diff --git a/src/langbot/pkg/storage/providers/s3storage.py b/src/langbot/pkg/storage/providers/s3storage.py index 8729da494..308665971 100644 --- a/src/langbot/pkg/storage/providers/s3storage.py +++ b/src/langbot/pkg/storage/providers/s3storage.py @@ -101,22 +101,37 @@ class S3StorageProvider(provider.StorageProvider): async def load( self, key: str, + ) -> bytes: + return await self.load_bounded(key, max_bytes=provider.HARD_MAX_STORAGE_OBJECT_BYTES) + + async def load_bounded( + self, + key: str, + *, + max_bytes: int, ) -> bytes: """Load bytes from S3""" + max_bytes = provider.normalize_read_limit(max_bytes) try: - return await self._run_io(self._load_sync, key) + return await self._run_io(self._load_sync, key, max_bytes) except Exception as e: self.ap.logger.error(f'Failed to load from S3: {e}') raise - def _load_sync(self, key: str) -> bytes: + def _load_sync(self, key: str, max_bytes: int) -> bytes: response = self.s3_client.get_object( Bucket=self.bucket_name, Key=key, ) body = response['Body'] try: - return body.read() + declared_size = response.get('ContentLength') + if declared_size is not None and declared_size > max_bytes: + raise ValueError(f'Storage object exceeds the {max_bytes}-byte read limit') + value = body.read(max_bytes + 1) + if len(value) > max_bytes: + raise ValueError(f'Storage object exceeds the {max_bytes}-byte read limit') + return value finally: body.close() diff --git a/src/langbot/pkg/vector/vdbs/valkey_search.py b/src/langbot/pkg/vector/vdbs/valkey_search.py index 6f96fccb9..48cf27a4c 100644 --- a/src/langbot/pkg/vector/vdbs/valkey_search.py +++ b/src/langbot/pkg/vector/vdbs/valkey_search.py @@ -82,6 +82,7 @@ _MATCH_ALL = '-@file_id:{__langbot_match_all_sentinel__}' # files/filters matching more than one page of chunks are fully removed # (no silent truncation / orphaned vectors). _DELETE_SCAN_BATCH = 10000 +_MAX_DELETE_SCAN_ROUNDS = 1000 # Characters Valkey Search's TAG query parser cannot handle even when # backslash-escaped (the brace delimiters and the wildcard). file_id TAG @@ -652,11 +653,9 @@ class ValkeySearchVectorDatabase(VectorDatabase): return query = f'@{_FIELD_FILE_ID}:{{{self._encode_and_escape_tag(file_id)}}}' - keys = await self._search_keys(client, index, query) - if keys: - await client.delete(keys) + deleted = await self._delete_search_results(client, index, query) self.ap.logger.info( - f"Deleted {len(keys)} embeddings from Valkey Search collection '{collection}' with file_id: {file_id}" + f"Deleted {deleted} embeddings from Valkey Search collection '{collection}' with file_id: {file_id}" ) async def delete_by_filter(self, collection: str, filter: dict[str, Any]) -> int: @@ -676,11 +675,9 @@ class ValkeySearchVectorDatabase(VectorDatabase): collection, ) return 0 - keys = await self._search_keys(client, index, query) - if keys: - await client.delete(keys) - self.ap.logger.info(f"Deleted {len(keys)} embeddings from Valkey Search collection '{collection}' by filter") - return len(keys) + deleted = await self._delete_search_results(client, index, query) + self.ap.logger.info(f"Deleted {deleted} embeddings from Valkey Search collection '{collection}' by filter") + return deleted async def list_by_filter( self, @@ -783,38 +780,30 @@ class ValkeySearchVectorDatabase(VectorDatabase): except RequestError: return False - async def _search_keys(self, client: GlideClient, index: str, query: str) -> list[str]: - """Return all matching document keys for a query (NOCONTENT). + async def _delete_search_results(self, client: GlideClient, index: str, query: str) -> int: + """Delete matching hashes in fixed batches without retaining every key. - Paginates through the full result set in pages of ``_DELETE_SCAN_BATCH`` - so that queries matching more than one page of chunks are fully - enumerated (avoids silently truncating deletes and leaving orphaned - vectors). + Each deletion shrinks the result set, so every search starts at offset + zero. Advancing an offset after deleting the preceding page would skip + records as the remaining results shift left. """ - keys: list[str] = [] - offset = 0 - while True: + + deleted = 0 + for _round in range(_MAX_DELETE_SCAN_ROUNDS): options = FtSearchOptions( nocontent=True, - limit=FtSearchLimit(offset, _DELETE_SCAN_BATCH), + limit=FtSearchLimit(0, _DELETE_SCAN_BATCH), dialect=2, ) try: reply = await ft.search(client, index, query, options) except Exception as exc: if self._is_missing_index_error(exc): - return keys + return deleted raise if not reply or len(reply) < 2: - break - - # reply[0] is the total match count; reply[1] holds this page. - total = 0 - try: - total = int(reply[0]) - except (TypeError, ValueError): - total = 0 + return deleted docs = reply[1] if isinstance(docs, dict): @@ -825,11 +814,17 @@ class ValkeySearchVectorDatabase(VectorDatabase): page = [] if not page: - break - keys.extend(page) + return deleted + await client.delete(page) + deleted += len(page) - offset += len(page) - if offset >= total or len(page) < _DELETE_SCAN_BATCH: - break + try: + total = int(reply[0]) + except (TypeError, ValueError): + total = len(page) + if total <= len(page) or len(page) < _DELETE_SCAN_BATCH: + return deleted - return keys + raise RuntimeError( + f'Valkey deletion exceeded {_MAX_DELETE_SCAN_ROUNDS} batches ({_DELETE_SCAN_BATCH} keys per batch)' + ) diff --git a/src/langbot/templates/config.yaml b/src/langbot/templates/config.yaml index 30294d4f3..a5a34d739 100644 --- a/src/langbot/templates/config.yaml +++ b/src/langbot/templates/config.yaml @@ -50,6 +50,16 @@ concurrency: # Hard admission limits for queued + running pipeline queries. pending_queries: 1000 pending_queries_per_workspace: 100 +webhooks: + # Bound database materialization and per-message outbound fan-out. + # Existing rows above this limit remain deletable through the management + # API, but only this many enabled destinations are dispatched. + # Supports WEBHOOKS__MAX_PER_WORKSPACE (hard cap: 64). + max_per_workspace: 16 + # Instance-wide request admission. Delivery fails open when every slot is + # occupied instead of retaining an unbounded queue of webhook tasks. + # Supports WEBHOOKS__MAX_INFLIGHT_REQUESTS (hard cap: 128). + max_inflight_requests: 16 cloud: # Operational safety ceilings for the one logical Cloud instance. These # are not subscription entitlements. An authoritative directory update @@ -208,6 +218,9 @@ vdb: request_timeout: 5000 # per-request timeout in ms (glide default 250ms is too low for KNN) storage: use: local + # Bound every object materialized into Core memory. Built-in Local/S3 + # providers enforce this while reading (hard cap: 64 MiB). + max_object_read_bytes: 10485760 cleanup: # Enable periodic cleanup of local/S3 uploaded files and old log files enabled: true @@ -319,6 +332,10 @@ box: max_sessions: 64 max_managed_processes: 64 max_completed_processes: 256 + # Core scans a Workspace before and after quota-enforced executions. + # Fail closed instead of repeatedly walking an inode bomb. + # Supports BOX__LIMITS__MAX_WORKSPACE_ENTRIES (hard cap: 1000000). + max_workspace_entries: 100000 # Retained admission fences prevent replay after entitlement expiry or # revocation. Fail closed before that monotonic state can grow without # bound; Cloud may override this with BOX__LIMITS__MAX_ADMISSION_RECORDS. diff --git a/tests/unit_tests/api/service/test_webhook_service.py b/tests/unit_tests/api/service/test_webhook_service.py index 0948c123b..64f58de05 100644 --- a/tests/unit_tests/api/service/test_webhook_service.py +++ b/tests/unit_tests/api/service/test_webhook_service.py @@ -50,11 +50,12 @@ def _create_mock_webhook( return webhook -def _create_mock_result(items: list = None, first_item=None): +def _create_mock_result(items: list = None, first_item=None, scalar_value=None): """Create mock result object for persistence queries.""" result = Mock() result.all = Mock(return_value=items or []) result.first = Mock(return_value=first_item) + result.scalar = Mock(return_value=scalar_value) result.rowcount = 1 return result @@ -154,6 +155,8 @@ class TestWebhookServiceCreateWebhook: nonlocal call_count call_count += 1 if call_count == 1: + return _create_mock_result(scalar_value=0) # Count + if call_count == 2: return insert_result # Insert return select_result # Select @@ -205,6 +208,8 @@ class TestWebhookServiceCreateWebhook: nonlocal call_count call_count += 1 if call_count == 1: + return _create_mock_result(scalar_value=0) + if call_count == 2: return _create_write_result() # Insert return _create_mock_result(first_item=created_webhook) @@ -246,6 +251,8 @@ class TestWebhookServiceCreateWebhook: nonlocal call_count call_count += 1 if call_count == 1: + return _create_mock_result(scalar_value=0) + if call_count == 2: return _create_write_result() return _create_mock_result(first_item=created_webhook) @@ -265,6 +272,42 @@ class TestWebhookServiceCreateWebhook: # Verify assert result['enabled'] is False + async def test_create_webhook_rejects_workspace_at_capacity(self): + ap = SimpleNamespace( + instance_config=SimpleNamespace( + data={'webhooks': {'max_per_workspace': 2}}, + ), + persistence_mgr=SimpleNamespace( + execute_async=AsyncMock(return_value=_create_mock_result(scalar_value=2)), + ), + ) + + service = WebhookService(ap) + + with pytest.raises(ValueError, match=r'Maximum number of webhooks \(2\) reached'): + await service.create_webhook( + WORKSPACE_UUID, + name='Too many', + url='https://example.invalid', + ) + + ap.persistence_mgr.execute_async.assert_awaited_once() + + async def test_max_per_workspace_clamps_invalid_and_oversized_values(self): + ap = SimpleNamespace( + instance_config=SimpleNamespace( + data={'webhooks': {'max_per_workspace': 999999}}, + ) + ) + service = WebhookService(ap) + assert service.max_per_workspace() == 64 + + ap.instance_config.data['webhooks']['max_per_workspace'] = 0 + assert service.max_per_workspace() == 1 + + ap.instance_config.data['webhooks']['max_per_workspace'] = 'invalid' + assert service.max_per_workspace() == 16 + class TestWebhookServiceGetWebhook: """Tests for get_webhook method.""" diff --git a/tests/unit_tests/box/test_box_service.py b/tests/unit_tests/box/test_box_service.py index bce4a03e5..9c78f771c 100644 --- a/tests/unit_tests/box/test_box_service.py +++ b/tests/unit_tests/box/test_box_service.py @@ -1280,6 +1280,45 @@ async def test_box_service_rejects_and_cleans_up_when_execution_exceeds_workspac assert backend.stop_calls == ['person_test_user'] +@pytest.mark.asyncio +async def test_box_service_rejects_workspace_inode_bomb_before_execution(tmp_path): + logger = Mock() + backend = FakeBackend(logger) + runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300) + host_dir = tmp_path / 'quota-workspace-entries' + host_dir.mkdir() + app = make_app(logger, [str(tmp_path)], workspace_quota_mb=1) + app.instance_config.data['box']['local']['default_workspace'] = str(host_dir) + app.instance_config.data['box']['limits'] = {'max_workspace_entries': 2} + service = BoxService(app, client=_InProcessBoxRuntimeClient(logger, runtime)) + + tenant_host_dir = service._tenant_workspace(_CONTEXT) + assert tenant_host_dir is not None + os.makedirs(tenant_host_dir, exist_ok=True) + for index in range(3): + pathlib.Path(tenant_host_dir, f'tiny-{index}').write_bytes(b'x') + + await service.initialize() + + with pytest.raises(BoxValidationError, match='workspace entry limit exceeded before execution'): + await service.execute_tool({'command': 'echo hi'}, make_query(46)) + + assert backend.start_calls == [] + + +def test_box_service_workspace_entry_limit_is_hard_clamped(): + app = make_app(Mock()) + app.instance_config.data['box']['limits'] = {'max_workspace_entries': 10_000_000} + service = BoxService(app, client=Mock(spec=BoxRuntimeClient)) + assert service._max_workspace_entries() == 1_000_000 + + app.instance_config.data['box']['limits']['max_workspace_entries'] = 0 + assert service._max_workspace_entries() == 1 + + app.instance_config.data['box']['limits']['max_workspace_entries'] = 'invalid' + assert service._max_workspace_entries() == 100_000 + + @pytest.mark.asyncio async def test_profile_offline_readonly_locks_read_only_rootfs(): """offline_readonly locks read_only_rootfs so it cannot be overridden.""" @@ -2126,7 +2165,7 @@ class TestInboundOutboundRoundTrip: async def fake_execute_tool(parameters, q): calls.append(parameters['command']) - if 'os.walk' in parameters['command']: + if 'os.scandir' in parameters['command']: return { 'ok': True, 'stdout': '[{"name": "out.png", "b64": "QUJD"}]', @@ -2156,7 +2195,7 @@ class TestInboundOutboundRoundTrip: async def fake_execute_tool(parameters, q): calls.append(parameters['command']) - if 'os.walk' in parameters['command']: + if 'os.scandir' in parameters['command']: return {'ok': True, 'stdout': '[]', 'stderr': ''} return {'ok': True, 'stdout': '', 'stderr': ''} diff --git a/tests/unit_tests/box/test_workspace.py b/tests/unit_tests/box/test_workspace.py index 1e41fb757..48720a48a 100644 --- a/tests/unit_tests/box/test_workspace.py +++ b/tests/unit_tests/box/test_workspace.py @@ -65,6 +65,9 @@ def test_wrap_python_command_with_env_contains_bootstrap_and_command(): assert '_LB_SYSTEM_PYTHON="$(command -v python3 || command -v python || true)"' in command assert '"$_LB_SYSTEM_PYTHON" -m venv "$_LB_VENV_DIR"' in command assert 'kill -0 "$_LB_LOCK_OWNER"' in command + assert 'max_manifest_bytes = 10 * 1024 * 1024' in command + assert 'handle.read(1024 * 1024)' in command + assert 'digest.update(handle.read())' not in command assert 'export VIRTUAL_ENV="$_LB_VENV_DIR"' in command assert command.rstrip().endswith('python script.py') diff --git a/tests/unit_tests/pipeline/test_longtext_image.py b/tests/unit_tests/pipeline/test_longtext_image.py index c02c8ecef..f0d703e8e 100644 --- a/tests/unit_tests/pipeline/test_longtext_image.py +++ b/tests/unit_tests/pipeline/test_longtext_image.py @@ -1,6 +1,11 @@ +from types import SimpleNamespace from unittest.mock import Mock +import pytest + +import langbot_plugin.api.entities.builtin.platform.message as platform_message from langbot.pkg.pipeline.longtext.strategies.image import Text2ImageStrategy +from langbot.pkg.pipeline.longtext.strategies import image class _WideFont: @@ -15,3 +20,31 @@ def test_image_strategy_line_split_always_consumes_input(): assert lines == ['a', 'b', 'c'] assert ''.join(lines) == 'abc' + + +def test_image_strategy_numeric_boundaries_are_found_in_linear_order(): + strategy = Text2ImageStrategy(Mock()) + + assert strategy.indexNumber('a12-b12-c345') == [['12', 1], ['12', 5], ['345', 9]] + + +def test_image_strategy_rejects_unbounded_line_count_before_allocating_canvas(monkeypatch): + strategy = Text2ImageStrategy(Mock()) + monkeypatch.setattr(image, '_MAX_TEXT_TO_IMAGE_LINES', 2) + + with pytest.raises(ValueError, match='2 lines'): + strategy._split_text_lines('one\ntwo\nthree', 1000, _WideFont()) + + +@pytest.mark.asyncio +async def test_image_strategy_falls_back_to_forward_for_oversized_text(monkeypatch): + app = Mock() + strategy = Text2ImageStrategy(app) + monkeypatch.setattr(image, '_MAX_TEXT_TO_IMAGE_CHARS', 4) + query = SimpleNamespace(adapter=SimpleNamespace(bot_account_id='bot')) + + components = await strategy.process('12345', query) + + assert len(components) == 1 + assert isinstance(components[0], platform_message.Forward) + app.logger.warning.assert_called_once() diff --git a/tests/unit_tests/platform/test_webhook_pusher.py b/tests/unit_tests/platform/test_webhook_pusher.py new file mode 100644 index 000000000..696858a03 --- /dev/null +++ b/tests/unit_tests/platform/test_webhook_pusher.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import asyncio +import logging +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from langbot.pkg.platform.webhook_pusher import WebhookPusher + + +pytestmark = pytest.mark.asyncio + + +def _application(max_inflight_requests: object) -> SimpleNamespace: + return SimpleNamespace( + instance_config=SimpleNamespace( + data={ + 'webhooks': { + 'max_inflight_requests': max_inflight_requests, + } + } + ), + logger=logging.getLogger(__name__), + ) + + +async def test_delivery_admission_never_queues_above_instance_limit(): + pusher = WebhookPusher(_application(2)) + release = asyncio.Event() + both_started = asyncio.Event() + calls = 0 + active = 0 + peak_active = 0 + + async def fake_push(url: str, payload: dict) -> dict: + nonlocal calls, active, peak_active + calls += 1 + active += 1 + peak_active = max(peak_active, active) + if active == 2: + both_started.set() + try: + await release.wait() + return {'url': url} + finally: + active -= 1 + + pusher._push_to_webhook = fake_push + webhooks = [{'url': f'https://example.invalid/{index}'} for index in range(5)] + + first_delivery = asyncio.create_task(pusher._push_to_webhooks(webhooks, {})) + await asyncio.wait_for(both_started.wait(), timeout=1) + second_results = await pusher._push_to_webhooks(webhooks, {}) + release.set() + first_results = await first_delivery + + assert len(first_results) == 2 + assert second_results == [] + assert calls == 2 + assert peak_active == 2 + assert pusher._inflight_requests == 0 + + +async def test_cancelled_delivery_reaps_children_and_releases_slots(): + pusher = WebhookPusher(_application(1)) + started = asyncio.Event() + never = asyncio.Event() + + async def blocking_push(url: str, payload: dict) -> dict: + started.set() + await never.wait() + return {} + + pusher._push_to_webhook = blocking_push + delivery = asyncio.create_task( + pusher._push_to_webhooks([{'url': 'https://example.invalid'}], {}), + ) + await asyncio.wait_for(started.wait(), timeout=1) + + delivery.cancel() + with pytest.raises(asyncio.CancelledError): + await delivery + + assert pusher._inflight_requests == 0 + pusher._push_to_webhook = AsyncMock(return_value={}) + assert await pusher._push_to_webhooks([{'url': 'https://example.invalid'}], {}) == [{}] + + +async def test_max_inflight_requests_clamps_config(): + pusher = WebhookPusher(_application(999999)) + assert pusher._max_inflight_requests() == 128 + + pusher.ap.instance_config.data['webhooks']['max_inflight_requests'] = 0 + assert pusher._max_inflight_requests() == 1 + + pusher.ap.instance_config.data['webhooks']['max_inflight_requests'] = 'invalid' + assert pusher._max_inflight_requests() == 16 diff --git a/tests/unit_tests/platform/test_websocket_adapter_attachments.py b/tests/unit_tests/platform/test_websocket_adapter_attachments.py index 438e4c5c6..35a6bb644 100644 --- a/tests/unit_tests/platform/test_websocket_adapter_attachments.py +++ b/tests/unit_tests/platform/test_websocket_adapter_attachments.py @@ -37,6 +37,7 @@ def _make_adapter(load_return=b'hello', load_side_effect=None): provider.load = AsyncMock(return_value=load_return, side_effect=load_side_effect) storage_mgr = Mock() storage_mgr.storage_provider = provider + storage_mgr.load_scoped_object_key = AsyncMock(return_value=load_return, side_effect=load_side_effect) storage_mgr.scoped_prefix.return_value = _UPLOAD_PREFIX storage_mgr.is_scoped_object_key.return_value = True storage_mgr.delete_scoped_object_key = AsyncMock() @@ -94,7 +95,7 @@ async def test_file_uses_octet_stream_fallback(): @pytest.mark.asyncio async def test_skips_components_without_path_or_unknown_type(): - adapter, _, provider = _make_adapter() + adapter, storage_mgr, provider = _make_adapter() chain = [ {'type': 'Image', 'path': ''}, # no path {'type': 'Plain', 'path': 'storage://abc/x'}, # not a file component @@ -102,6 +103,7 @@ async def test_skips_components_without_path_or_unknown_type(): ] await adapter._process_image_components(_make_connection(), chain) provider.load.assert_not_awaited() + storage_mgr.load_scoped_object_key.assert_not_awaited() assert 'base64' not in chain[0] assert 'base64' not in chain[1] diff --git a/tests/unit_tests/platform/test_websocket_session_isolation.py b/tests/unit_tests/platform/test_websocket_session_isolation.py index 240e7acfc..682a42184 100644 --- a/tests/unit_tests/platform/test_websocket_session_isolation.py +++ b/tests/unit_tests/platform/test_websocket_session_isolation.py @@ -327,7 +327,7 @@ async def test_attachment_key_must_belong_to_connection_upload_scope(): storage_mgr = Mock() storage_mgr.scoped_prefix.return_value = 'v1/current/upload_image/' storage_mgr.is_scoped_object_key.return_value = True - storage_mgr.storage_provider.load = AsyncMock(return_value=b'image') + storage_mgr.load_scoped_object_key = AsyncMock(return_value=b'image') storage_mgr.delete_scoped_object_key = AsyncMock() adapter = WebSocketAdapter.model_construct( ap=Mock(storage_mgr=storage_mgr), @@ -347,6 +347,11 @@ async def test_attachment_key_must_belong_to_connection_upload_scope(): 'v1/current/upload_image/key.png', expected_owner_type='upload_image', ) + storage_mgr.load_scoped_object_key.assert_awaited_once_with( + connection.execution_context, + 'v1/current/upload_image/key.png', + expected_owner_type='upload_image', + ) storage_mgr.delete_scoped_object_key.assert_awaited_once_with( connection.execution_context, 'v1/current/upload_image/key.png', diff --git a/tests/unit_tests/plugin/test_handler_actions.py b/tests/unit_tests/plugin/test_handler_actions.py index c215670c1..4089e23be 100644 --- a/tests/unit_tests/plugin/test_handler_actions.py +++ b/tests/unit_tests/plugin/test_handler_actions.py @@ -417,7 +417,9 @@ class TestGetConfigFile: mock_app.persistence_mgr = Mock() mock_app.persistence_mgr.execute_async = AsyncMock() mock_app.storage_mgr = StorageMgr(mock_app) - mock_app.storage_mgr.storage_provider = SimpleNamespace(load=AsyncMock(return_value=b'plugin config bytes')) + mock_app.storage_mgr.storage_provider = SimpleNamespace( + load_bounded=AsyncMock(return_value=b'plugin config bytes') + ) mock_app.logger = Mock() return mock_app @@ -461,7 +463,10 @@ class TestGetConfigFile: assert response.code == 0 assert base64.b64decode(response.data['file_base64']) == b'plugin config bytes' - app.storage_mgr.storage_provider.load.assert_awaited_once_with(file_key) + app.storage_mgr.storage_provider.load_bounded.assert_awaited_once_with( + file_key, + max_bytes=10 * 1024 * 1024, + ) @pytest.mark.asyncio @pytest.mark.parametrize( @@ -478,7 +483,7 @@ class TestGetConfigFile: assert response.code != 0 assert 'Failed to load config file' in response.message - app.storage_mgr.storage_provider.load.assert_not_awaited() + app.storage_mgr.storage_provider.load_bounded.assert_not_awaited() class TestGetBinaryStorage: diff --git a/tests/unit_tests/provider/test_tool_manager_native.py b/tests/unit_tests/provider/test_tool_manager_native.py index d18a830dc..ae50e3935 100644 --- a/tests/unit_tests/provider/test_tool_manager_native.py +++ b/tests/unit_tests/provider/test_tool_manager_native.py @@ -584,6 +584,40 @@ async def test_host_file_api_falls_back_to_tenant_box_when_openat_is_unavailable assert 'path = "/workspace/file.txt"' in command +@pytest.mark.asyncio +async def test_box_workspace_edit_script_bounds_file_read_and_replacement(monkeypatch): + with tempfile.TemporaryDirectory() as tmpdir: + box_service = SimpleNamespace( + available=True, + default_workspace=tmpdir, + _tenant_workspace=Mock(return_value=tmpdir), + execute_tool=AsyncMock( + return_value={ + 'ok': True, + 'stdout': '{"ok": false, "error": "File exceeds limit"}', + 'stderr': '', + } + ), + ) + loader = NativeToolLoader(SimpleNamespace(box_service=box_service, logger=Mock())) + monkeypatch.setattr(native_loader, '_SECURE_HOST_FILE_OPS_AVAILABLE', False) + + await loader.invoke_tool( + 'edit', + { + 'path': '/workspace/file.txt', + 'old_string': 'old', + 'new_string': 'new', + }, + _make_query(), + ) + + command = box_service.execute_tool.await_args.args[0]['command'] + assert f'os.path.getsize(path) > {native_loader._MAX_HOST_EDIT_FILE_BYTES}' in command + assert f'f.read({native_loader._MAX_HOST_EDIT_FILE_BYTES + 1})' in command + assert f"len(new_content.encode('utf-8')) > {native_loader._MAX_HOST_EDIT_FILE_BYTES}" in command + + @pytest.mark.asyncio async def test_box_availability_helper_handles_unavailable_and_errors(): from langbot.pkg.provider.tools.loaders.availability import is_box_backend_available diff --git a/tests/unit_tests/rag/test_file_storage.py b/tests/unit_tests/rag/test_file_storage.py index 9adec28f0..7b464d502 100644 --- a/tests/unit_tests/rag/test_file_storage.py +++ b/tests/unit_tests/rag/test_file_storage.py @@ -52,6 +52,7 @@ def _make_app() -> Mock: storage_mgr.storage_provider = Mock() storage_mgr.storage_provider.exists = AsyncMock(return_value=True) storage_mgr.storage_provider.load = AsyncMock() + storage_mgr.storage_provider.load_bounded = AsyncMock() storage_mgr.storage_provider.save = AsyncMock() storage_mgr.storage_provider.size = AsyncMock(return_value=123) storage_mgr.storage_provider.delete = AsyncMock() @@ -175,7 +176,7 @@ class TestStoreZipFile: @pytest.mark.asyncio async def test_store_zip_file_extracts_supported_files_and_skips_noise(self): kb = _make_kb() - kb.ap.storage_mgr.storage_provider.load = AsyncMock( + kb.ap.storage_mgr.storage_provider.load_bounded = AsyncMock( return_value=_make_zip_bytes( { 'doc1.pdf': b'pdf', @@ -210,7 +211,7 @@ class TestStoreZipFile: @pytest.mark.asyncio async def test_store_zip_file_raises_when_no_supported_files(self): kb = _make_kb() - kb.ap.storage_mgr.storage_provider.load = AsyncMock( + kb.ap.storage_mgr.storage_provider.load_bounded = AsyncMock( return_value=_make_zip_bytes({'image.png': b'png', 'video.mp4': b'video'}) ) kb.store_file = AsyncMock() @@ -224,7 +225,7 @@ class TestStoreZipFile: @pytest.mark.asyncio async def test_store_zip_file_rejects_too_many_documents_before_extracting(self): kb = _make_kb() - kb.ap.storage_mgr.storage_provider.load = AsyncMock( + kb.ap.storage_mgr.storage_provider.load_bounded = AsyncMock( return_value=_make_zip_bytes({f'doc-{index}.txt': b'text' for index in range(9)}) ) kb.store_file = AsyncMock() @@ -239,7 +240,7 @@ class TestStoreZipFile: @pytest.mark.asyncio async def test_store_zip_file_rejects_extreme_compression_ratio_before_extracting(self): kb = _make_kb() - kb.ap.storage_mgr.storage_provider.load = AsyncMock( + kb.ap.storage_mgr.storage_provider.load_bounded = AsyncMock( return_value=_make_zip_bytes( {'bomb.txt': b'A' * (1024 * 1024)}, compression=zipfile.ZIP_DEFLATED, diff --git a/tests/unit_tests/storage/test_localstorage_path_traversal.py b/tests/unit_tests/storage/test_localstorage_path_traversal.py index 5e950eb32..dc72da5d1 100644 --- a/tests/unit_tests/storage/test_localstorage_path_traversal.py +++ b/tests/unit_tests/storage/test_localstorage_path_traversal.py @@ -168,6 +168,15 @@ class TestPathTraversalPrevention: assert loaded == content await provider.delete(key) + @pytest.mark.asyncio + async def test_bounded_load_stops_after_limit(self, storage_provider): + provider, storage_path = storage_provider + + with patch('langbot.pkg.storage.providers.localstorage.LOCAL_STORAGE_PATH', storage_path): + await provider.save('oversized.bin', b'12345') + with pytest.raises(ValueError, match='4-byte read limit'): + await provider.load_bounded('oversized.bin', max_bytes=4) + @pytest.mark.asyncio async def test_delete_dir_recursive_non_existing_dir(self, storage_provider): """delete_dir_recursive should handle non-existing directories gracefully.""" diff --git a/tests/unit_tests/storage/test_s3storage.py b/tests/unit_tests/storage/test_s3storage.py index a325e5c61..7e90cf43d 100644 --- a/tests/unit_tests/storage/test_s3storage.py +++ b/tests/unit_tests/storage/test_s3storage.py @@ -134,6 +134,16 @@ class TestS3StorageProviderWithMoto: loaded_data = await provider.load('test/file.txt') assert loaded_data == test_data + @pytest.mark.asyncio + async def test_bounded_load_rejects_oversized_object(self, mock_app_with_s3_config, s3_mock): + s3storage = get_s3storage_module() + provider = s3storage.S3StorageProvider(mock_app_with_s3_config) + await provider.initialize() + await provider.save('test/oversized.bin', b'12345') + + with pytest.raises(ValueError, match='4-byte read limit'): + await provider.load_bounded('test/oversized.bin', max_bytes=4) + @pytest.mark.asyncio async def test_exists_returns_true_for_existing_object(self, mock_app_with_s3_config, s3_mock): """Test that exists returns True for existing object.""" diff --git a/tests/unit_tests/storage/test_workspace_scoping.py b/tests/unit_tests/storage/test_workspace_scoping.py index 79869a85c..2787ba88b 100644 --- a/tests/unit_tests/storage/test_workspace_scoping.py +++ b/tests/unit_tests/storage/test_workspace_scoping.py @@ -8,6 +8,7 @@ import pytest from langbot.pkg.api.http.authz import WorkspaceRequiredError from langbot.pkg.api.http.context import ExecutionContext from langbot.pkg.storage.mgr import StorageMgr +from langbot.pkg.storage.provider import HARD_MAX_STORAGE_OBJECT_BYTES from langbot.pkg.utils.bounded_executor import current_blocking_work_scope @@ -202,6 +203,39 @@ async def test_opaque_object_operations_reject_cross_scope_and_owner_type(manage assert object_key in manager.storage_provider.values +@pytest.mark.asyncio +async def test_scoped_object_reads_and_writes_have_configured_and_hard_byte_limits(manager): + manager.ap.instance_config = SimpleNamespace(data={'storage': {'max_object_read_bytes': 4}}) + + with pytest.raises(ValueError, match='4-byte write limit'): + await manager.save_scoped( + _context(WORKSPACE_A), + owner_type='upload', + owner='account:a', + key='oversized.bin', + value=b'12345', + ) + + object_key = manager.scoped_object_key( + _context(WORKSPACE_A), + owner_type='upload', + owner='account:a', + key='external.bin', + ) + manager.storage_provider.values[object_key] = b'12345' + manager.storage_provider.load = AsyncMock(side_effect=AssertionError('oversized object must not be loaded')) + with pytest.raises(ValueError, match='4-byte read limit'): + await manager.load_scoped_object_key( + _context(WORKSPACE_A), + object_key, + expected_owner_type='upload', + ) + manager.storage_provider.load.assert_not_awaited() + + manager.ap.instance_config.data['storage']['max_object_read_bytes'] = HARD_MAX_STORAGE_OBJECT_BYTES + 1 + assert manager._object_read_limit() == HARD_MAX_STORAGE_OBJECT_BYTES + + @pytest.mark.asyncio async def test_scoped_provider_is_not_touched_after_generation_is_fenced(manager): object_key = await manager.save_scoped( diff --git a/tests/unit_tests/vector/test_valkey_search_filter.py b/tests/unit_tests/vector/test_valkey_search_filter.py index 19c5bf138..c64c483cb 100644 --- a/tests/unit_tests/vector/test_valkey_search_filter.py +++ b/tests/unit_tests/vector/test_valkey_search_filter.py @@ -268,9 +268,9 @@ class TestDeleteByFilterGuard: backend.ap = type('Ap', (), {'logger': AsyncMock()})() backend._ensure_client = AsyncMock(return_value=backend._client) backend._index_exists = AsyncMock(return_value=True) - # _search_keys must never be reached for an unusable filter. - backend._search_keys = AsyncMock( - side_effect=AssertionError('_search_keys must not be called for an unusable filter') + # The deletion scan must never be reached for an unusable filter. + backend._delete_search_results = AsyncMock( + side_effect=AssertionError('_delete_search_results must not be called for an unusable filter') ) # Filter references only a non-indexed field -> maps to no FT conditions. @@ -285,12 +285,57 @@ class TestDeleteByFilterGuard: backend.ap = type('Ap', (), {'logger': AsyncMock()})() backend._ensure_client = AsyncMock(return_value=backend._client) backend._index_exists = AsyncMock(return_value=True) - backend._search_keys = AsyncMock(return_value=['kb:col1:id1', 'kb:col1:id2']) + backend._delete_search_results = AsyncMock(return_value=2) deleted = await backend.delete_by_filter('col1', {'file_id': 'f1'}) assert deleted == 2 - backend._client.delete.assert_awaited_once_with(['kb:col1:id1', 'kb:col1:id2']) + backend._delete_search_results.assert_awaited_once_with( + backend._client, + backend._index_name('col1'), + '@file_id:{f1}', + ) + + +class TestBatchedDelete: + async def test_matching_keys_are_deleted_in_fixed_pages(self, monkeypatch): + mod = get_valkey_module() + backend = make_backend() + client = AsyncMock() + search = AsyncMock( + side_effect=[ + [3, {b'key-1': {}, b'key-2': {}}], + [1, {b'key-3': {}}], + ] + ) + monkeypatch.setattr(mod, '_DELETE_SCAN_BATCH', 2) + monkeypatch.setattr(mod, 'FtSearchLimit', lambda offset, limit: (offset, limit), raising=False) + monkeypatch.setattr(mod, 'FtSearchOptions', lambda **kwargs: kwargs, raising=False) + monkeypatch.setattr(mod, 'ft', type('FT', (), {'search': search})(), raising=False) + + deleted = await backend._delete_search_results(client, 'idx:col1', '@file_id:{f1}') + + assert deleted == 3 + assert search.await_count == 2 + assert [call.args[3]['limit'] for call in search.await_args_list] == [(0, 2), (0, 2)] + assert client.delete.await_args_list[0].args == (['key-1', 'key-2'],) + assert client.delete.await_args_list[1].args == (['key-3'],) + + async def test_delete_rounds_have_a_hard_stop(self, monkeypatch): + mod = get_valkey_module() + backend = make_backend() + client = AsyncMock() + search = AsyncMock(return_value=[2, {b'key': {}}]) + monkeypatch.setattr(mod, '_DELETE_SCAN_BATCH', 1) + monkeypatch.setattr(mod, '_MAX_DELETE_SCAN_ROUNDS', 2) + monkeypatch.setattr(mod, 'FtSearchLimit', lambda offset, limit: (offset, limit), raising=False) + monkeypatch.setattr(mod, 'FtSearchOptions', lambda **kwargs: kwargs, raising=False) + monkeypatch.setattr(mod, 'ft', type('FT', (), {'search': search})(), raising=False) + + with pytest.raises(RuntimeError, match='exceeded 2 batches'): + await backend._delete_search_results(client, 'idx:col1', '@file_id:{f1}') + + assert client.delete.await_count == 2 class TestClose: diff --git a/uv.lock b/uv.lock index c233e1d3c..2c4bdb5c0 100644 --- a/uv.lock +++ b/uv.lock @@ -2116,7 +2116,7 @@ requires-dist = [ { name = "ebooklib", specifier = ">=0.18" }, { name = "gewechat-client", specifier = ">=0.1.5" }, { name = "html2text", specifier = ">=2024.2.26" }, - { name = "langbot-plugin", git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=7c0b9827ed8597a1c84151b83fcf6307934fd944" }, + { name = "langbot-plugin", git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=87d903e9b7c4e518ce066c89f65bbde4de825fc5" }, { name = "langchain", specifier = ">=1.3.9" }, { name = "langchain-core", specifier = ">=1.3.3" }, { name = "langchain-text-splitters", specifier = ">=1.1.2" }, @@ -2183,7 +2183,7 @@ dev = [ [[package]] name = "langbot-plugin" version = "0.4.18" -source = { git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=7c0b9827ed8597a1c84151b83fcf6307934fd944#7c0b9827ed8597a1c84151b83fcf6307934fd944" } +source = { git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=87d903e9b7c4e518ce066c89f65bbde4de825fc5#87d903e9b7c4e518ce066c89f65bbde4de825fc5" } dependencies = [ { name = "aiofiles" }, { name = "aiohttp" },