From fc951877105f918da9f693b4aa8c740990cad688 Mon Sep 17 00:00:00 2001 From: Shawn Wen <45581750+KOP-XIAO@users.noreply.github.com> Date: Sat, 29 Aug 2026 12:26:51 +0800 Subject: [PATCH 1/2] feat: add dsh (DeepSeek Harness) CLI backend Add dsh as a fourth CLI backend (agy/grok/codex/dsh). Runs `dsh --profile headless ` in the per-user session directory, extracts file:/// and markdown file links as send-back artifacts, and maps CLI errors to the standard Chinese notice bubbles. The headless profile always creates a fresh session per invocation, so the dsh backend is single-turn: /clear and /new are accepted but are no-ops, and model/effort/persona slash commands are not wired yet. - config.py: DSH_BIN_PATH / DSH_PROFILE / DSH_TIMEOUT / WECHATBRIDGE_DSH_HOME + backend whitelist - runner_common.py: register dsh in KNOWN_BACKENDS; .dsh cleanup paths - main.py: dispatch in _run_llm/_handle_slash/send_artifacts_back; /backend usage text now lists dsh - docs: README (en/zh), deploy env example, CHANGELOG - tests: tests/fake_dsh.py + tests/test_dsh.py (35 tests, all pass) --- CHANGELOG.md | 8 + README.md | 37 +++- README.zh-CN.md | 37 +++- deploy/wechatbridge.env.example | 18 +- tests/fake_dsh.py | 112 ++++++++++ tests/test_dsh.py | 379 ++++++++++++++++++++++++++++++++ wechatbridge/config.py | 19 +- wechatbridge/dsh.py | 346 +++++++++++++++++++++++++++++ wechatbridge/main.py | 23 +- wechatbridge/runner_common.py | 11 +- 10 files changed, 961 insertions(+), 29 deletions(-) create mode 100755 tests/fake_dsh.py create mode 100644 tests/test_dsh.py create mode 100644 wechatbridge/dsh.py diff --git a/CHANGELOG.md b/CHANGELOG.md index db186b9..89a6111 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,14 @@ All notable changes follow [Keep a Changelog](https://keepachangelog.com/) and ## [Unreleased] +### Added + +- **dsh 后端**：新增 DeepSeek Harness CLI（`dsh`）作为第四个 CLI 后端（`/backend dsh`）。以 `dsh --profile headless ` 单轮运行，cwd 为每用户会话目录；从输出提取 `file:///` 与 markdown 文件链接作为可回传产物；`DSH_HOME` 机器级共享（可 `WECHATBRIDGE_DSH_HOME` 覆盖）；`/help`、`/clear`、`/new` 支持，模型/强度/人格类指令暂返回"暂不支持"。headless 每次新建会话，故为单轮模式。 + +### Tests + +- 新增 `tests/test_dsh.py` + `tests/fake_dsh.py`：命令构造、产物提取、成功/失败/超时路径、未登录预检、slash 指令、`/backend dsh` 切换。 + ## [1.4.9] - 2026-08-16 ### Fixed diff --git a/README.md b/README.md index e35ed32..126095c 100644 --- a/README.md +++ b/README.md @@ -5,21 +5,21 @@ ![license](https://img.shields.io/badge/license-MIT-blue.svg) ![python](https://img.shields.io/badge/python-3.10+-blue.svg) -WeChatBridge connects a WeChat bot to agentic coding CLIs (Google's agy / Antigravity, xAI's Grok Build, or OpenAI's Codex). From WeChat you can send text, images, files, and voice-as-text to the active CLI, get replies back, and receive certain generated files over the WeChat CDN. Switch backends per user with `/backend` — no restart. +WeChatBridge connects a WeChat bot to agentic coding CLIs (Google's agy / Antigravity, xAI's Grok Build, OpenAI's Codex, or DeepSeek Harness' dsh). From WeChat you can send text, images, files, and voice-as-text to the active CLI, get replies back, and receive certain generated files over the WeChat CDN. Switch backends per user with `/backend` — no restart. ``` -WeChat (phone) ⇄ iLink bot API ⇄ WeChatBridge ⇄ agy / grok / codex CLI +WeChat (phone) ⇄ iLink bot API ⇄ WeChatBridge ⇄ agy / grok / codex / dsh CLI (this project) (runs tools) ``` -The bridge process stays up and long-polls iLink. For prompts that go to a CLI, it spawns one `agy` or `grok` child (`-p` single-turn) and exits that child when done — the child does not stay resident. Many slash commands (`/help`, `/backend`, `/persona`, …) are handled inside the bridge and never start a CLI. Only artifacts the bridge can detect under the user's allowed session paths are pushed back via CDN. +The bridge process stays up and long-polls iLink. For prompts that go to a CLI, it spawns one `agy` / `grok` / `codex` / `dsh` child (single-turn) and exits that child when done — the child does not stay resident. Many slash commands (`/help`, `/backend`, `/persona`, …) are handled inside the bridge and never start a CLI. Only artifacts the bridge can detect under the user's allowed session paths are pushed back via CDN. ## Features -- Text, image, file, and voice (WeChat server-side transcription only) go to the **active** backend (`agy`, `grok`, or `codex`) +- Text, image, file, and voice (WeChat server-side transcription only) go to the **active** backend (`agy`, `grok`, `codex`, or `dsh`) - Detected CLI artifacts under the per-user allowed tree can be sent back (size-capped); not every file the CLI touches - Each WeChat user gets an isolated workspace; model / effort / mode are remembered **per backend** -- Runtime backend switch: `/backend agy`, `/backend grok`, or `/backend codex` (clears that backend's continuation state — the agy/grok continuation flag and the codex `thread_id`/resume state — so the next CLI turn starts a fresh session; history files on disk are not wiped immediately) +- Runtime backend switch: `/backend agy`, `/backend grok`, `/backend codex`, or `/backend dsh` (clears that backend's continuation state — the agy/grok continuation flag and the codex `thread_id`/resume state — so the next CLI turn starts a fresh session; history files on disk are not wiped immediately) - Slash commands for model, session reset, persona, and more (see below) - Dangerous-prompt gate: a **keyword list** of concrete destructive patterns asks for confirmation before run - Sender whitelist (`WECHATBRIDGE_ALLOWED_SENDERS`; empty = allow all) @@ -41,8 +41,16 @@ Default data paths expand from `~` (e.g. `~/.local/share/wechatbridge/ - **agy** (default) — Google Antigravity CLI - **grok** — xAI Grok Build CLI - **codex** — OpenAI Codex CLI +- **dsh** — DeepSeek Harness CLI (one-shot `headless` profile) -Per-user switch: `/backend agy`, `/backend grok`, or `/backend codex`. Each backend keeps its own model / effort / mode memory and persona file layout. Global default is `WECHATBRIDGE_BACKEND`. +Per-user switch: `/backend agy`, `/backend grok`, `/backend codex`, or `/backend dsh`. Each backend keeps its own model / effort / mode memory and persona file layout. Global default is `WECHATBRIDGE_BACKEND`. + +### dsh backend notes + +- **Single-turn:** the `headless` profile always creates a fresh session per invocation (`session-`), so every WeChat message starts a new dsh session. `/clear` / `/new` are accepted but are no-ops, and `/model`, `/fast`, `/planning`, `/persona`, `/add-dir` are not wired yet (they return a short "not supported" notice). +- Runs `dsh --profile headless ` with `cwd` = the per-user session directory (per-user workspace; model-created files land there and can be sent back via CDN). +- Auth / profiles are **machine-wide** (`~/.dsh`, same model as grok's machine-wide login): the child's `HOME` points at the per-user session dir, so the bridge always passes `DSH_HOME` explicitly. Override with `WECHATBRIDGE_DSH_HOME` (e.g. a dedicated service home; for per-user `DSH_HOME` isolation, pre-seed profiles there). `DSH_BIN_PATH`, `DSH_PROFILE`, and `DSH_TIMEOUT` are configurable. +- Status: implemented from the published `dsh` CLI contract (headless bundle source) plus a fake CLI in the test suite; final acceptance depends on a real `dsh login` + headless run. ### Grok backend notes @@ -62,7 +70,8 @@ Per-user switch: `/backend agy`, `/backend grok`, or `/backend codex`. Each back - **agy** on `PATH`, or set `AGY_BIN_PATH` - **and/or grok** on `PATH`, or set `GROK_BIN_PATH` - **and/or codex** on `PATH`, or set `CODEX_BIN_PATH` - - Antigravity is Google's terminal agentic coding CLI (successor to Gemini CLI). Grok Build is xAI's counterpart; Codex is OpenAI's terminal agentic coding CLI. + - **and/or dsh** on `PATH`, or set `DSH_BIN_PATH` (DeepSeek Harness, `dsh login` required) + - Antigravity is Google's terminal agentic coding CLI (successor to Gemini CLI). Grok Build is xAI's counterpart; Codex is OpenAI's terminal agentic coding CLI; dsh is DeepSeek Harness' CLI. - A WeChat account with a [ClawBot / iLink](https://ilinkai.weixin.qq.com) bot (QR bind on first run) - Python 3.10+ @@ -133,10 +142,14 @@ Key variables (all have defaults): | `AGY_BIN_PATH` | `agy` | path to the agy binary | | `GROK_BIN_PATH` | `grok` | path to the grok binary | | `CODEX_BIN_PATH` | `codex` | path to the codex binary | -| `WECHATBRIDGE_BACKEND` | `agy` | global default backend (`agy` / `grok` / `codex`; overridable per user via `/backend`) | +| `DSH_BIN_PATH` | `dsh` | path to the dsh binary | +| `DSH_PROFILE` | `headless` | dsh profile booted for one-shot tasks | +| `DSH_TIMEOUT` | `600` | dsh CLI run timeout in seconds | +| `WECHATBRIDGE_DSH_HOME` | _empty_ | explicit `DSH_HOME` passed to the dsh child (empty = machine-wide `~/.dsh`) | +| `WECHATBRIDGE_BACKEND` | `agy` | global default backend (`agy` / `grok` / `codex` / `dsh`; overridable per user via `/backend`) | | `WECHATBRIDGE_INSTANCE` | `default` | instance name; state / session / QR paths derive from it | | `WECHATBRIDGE_ALLOWED_SENDERS` | _empty_ | comma-separated WeChat IDs (empty = allow all) | -| `AGY_TIMEOUT` | `600` | CLI run timeout in seconds (all three backends) | +| `AGY_TIMEOUT` | `600` | CLI run timeout in seconds (agy / grok / codex backends) | | `WECHATBRIDGE_MAX_OUTBOUND_BYTES` | `104857600` | max file size sent back to WeChat (100 MB) | | `WECHATBRIDGE_MAX_INBOUND_BYTES` | `20971520` | max inbound image/file after download (20 MB) | | `WECHATBRIDGE_MAX_CONCURRENT` | `4` | global concurrent process slots; same user serial (queue does not hold a slot); extras get a busy reply | @@ -222,7 +235,7 @@ See [`deploy/wechatbridge-windows.md`](deploy/wechatbridge-windows.md). | Command | Action | |---|---| | `/help` | list supported commands for the active backend | -| `/backend ` | switch CLI backend for this WeChat user (on real change: clears that backend's continuation state — agy/grok flag and codex `thread_id`/resume — so the next turn starts a fresh session; history files may remain until retention cleanup) | +| `/backend ` | switch CLI backend for this WeChat user (on real change: clears that backend's continuation state — agy/grok flag and codex `thread_id`/resume — so the next turn starts a fresh session; history files may remain until retention cleanup) | | `/clear` or `/new` | drop continue flag so the next CLI turn is a new conversation (does not instantly delete history files) | | `/model ` | set model (all backends validate against a live list: agy/grok via CLI `models`; codex via `codex debug models` [then `--bundled`]; unknown name or list-fetch failure refuse and do not write prefs; see `/models`) | | `/models` | list models — agy/grok/codex all query the live CLI (codex: `debug models`; falls back to a built-in reference note only if the live list cannot be fetched) | @@ -254,8 +267,10 @@ Other `/…` commands are either rejected (e.g. `/exit`), reported as unsupporte ## Limitations -- Not a standalone agent — requires agy and/or grok and/or codex. +- Not a standalone agent — requires agy and/or grok and/or codex and/or dsh. - The **codex** backend is not yet verified against a real Codex subscription/CLI; it is validated by source research, a JSONL fixture, and a fake CLI in tests. Treat it as community-tested until a real user confirms. +- The **dsh** backend is single-turn only (the `headless` profile always starts a fresh session) and has not yet been verified against a real `dsh login` + headless run; it is validated against the published headless bundle contract and a fake CLI in tests. +- dsh model/effort/mode/persona slash commands are not wired yet; `/model`, `/fast`, `/planning`, `/persona`, `/add-dir` return a "not supported" notice on the dsh backend. - Voice is WeChat speech-to-text only; no local ASR; empty transcript → “type instead”. - No video send/receive; no native WeChat voice-bubble replies (no silk encode). - One WeChat binding per process; multiple accounts need multiple instances (`WECHATBRIDGE_INSTANCE`). diff --git a/README.zh-CN.md b/README.zh-CN.md index 1bed696..5a0f734 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -5,21 +5,21 @@ ![license](https://img.shields.io/badge/license-MIT-blue.svg) ![python](https://img.shields.io/badge/python-3.10+-blue.svg) -WeChatBridge 把微信机器人接到 agentic 编程 CLI（谷歌 agy / Antigravity、xAI Grok Build，或 OpenAI Codex）。你可以在微信里发文字、图片、文件、语音（仅微信侧转写文字）给**当前后端**，拿回复，并在条件满足时把部分生成文件经 CDN 发回微信。每个用户可用 `/backend` 切换后端，无需重启进程。 +WeChatBridge 把微信机器人接到 agentic 编程 CLI（谷歌 agy / Antigravity、xAI Grok Build、OpenAI Codex，或 DeepSeek Harness 的 dsh）。你可以在微信里发文字、图片、文件、语音（仅微信侧转写文字）给**当前后端**，拿回复，并在条件满足时把部分生成文件经 CDN 发回微信。每个用户可用 `/backend` 切换后端，无需重启进程。 ``` -微信(手机) ⇄ iLink 机器人 API ⇄ WeChatBridge ⇄ agy / grok / codex CLI +微信(手机) ⇄ iLink 机器人 API ⇄ WeChatBridge ⇄ agy / grok / codex / dsh CLI (本项目) (跑工具) ``` -桥接进程本身一直在跑、长轮询 iLink。需要交给 CLI 的提示才会按需拉起一个 agy 或 grok 子进程（`-p` 单轮），**子进程**跑完即退、不常驻。许多 slash（如 `/help`、`/backend`、`/persona`）在桥内直接处理，不会起 CLI。只有桥能识别、且落在用户允许目录内的产物，才会经 CDN 回传。 +桥接进程本身一直在跑、长轮询 iLink。需要交给 CLI 的提示才会按需拉起一个 agy / grok / codex / dsh 子进程（单轮），**子进程**跑完即退、不常驻。许多 slash（如 `/help`、`/backend`、`/persona`）在桥内直接处理，不会起 CLI。只有桥能识别、且落在用户允许目录内的产物，才会经 CDN 回传。 ## 功能 -- 文本、图片、文件、语音（仅微信服务端语音转文字）交给**当前激活**的后端（`agy`、`grok` 或 `codex`） +- 文本、图片、文件、语音（仅微信服务端语音转文字）交给**当前激活**的后端（`agy`、`grok`、`codex` 或 `dsh`） - 在用户允许目录内、且被识别到的 CLI 产物可回传微信（有大小上限）；不是 CLI 碰过的每个文件都会发 - 每个微信用户独立工作区；模型 / 推理强度 / 模式按**后端**分别记忆 -- 运行时切换后端：`/backend agy`、`/backend grok` 或 `/backend codex`（真正切换时清除该后端续聊状态——agy/grok 的续聊标记与 codex 的 `thread_id`/resume 状态——下次 CLI 起新会话；磁盘上的历史文件不会立刻抹掉） +- 运行时切换后端：`/backend agy`、`/backend grok`、`/backend codex` 或 `/backend dsh`（真正切换时清除该后端续聊状态——agy/grok 的续聊标记与 codex 的 `thread_id`/resume 状态——下次 CLI 起新会话；磁盘上的历史文件不会立刻抹掉） - slash 指令：模型、清会话、人格等（见下表） - 危险提示闸门：对**明确破坏性关键词/模式**先确认再跑（不是全语义理解） - 白名单 `WECHATBRIDGE_ALLOWED_SENDERS`（空 = 全开） @@ -41,8 +41,16 @@ WeChatBridge 把微信机器人接到 agentic 编程 CLI（谷歌 agy / Antigrav - **agy**（默认）— 谷歌 Antigravity CLI - **grok** — xAI Grok Build CLI - **codex** — OpenAI Codex CLI +- **dsh** — DeepSeek Harness CLI（单轮 `headless` profile） -微信里 `/backend agy`、`/backend grok` 或 `/backend codex` 按用户切换。各后端各自记模型 / 强度 / 模式，人格文件布局也分开。全局默认见 `WECHATBRIDGE_BACKEND`。 +微信里 `/backend agy`、`/backend grok`、`/backend codex` 或 `/backend dsh` 按用户切换。各后端各自记模型 / 强度 / 模式，人格文件布局也分开。全局默认见 `WECHATBRIDGE_BACKEND`。 + +### dsh 后端说明 + +- **单轮模式：** `headless` profile 每次调用都新建会话（`session-`），所以每条微信消息都会开启全新 dsh 会话。`/clear` / `/new` 接受但无实际作用；`/model`、`/fast`、`/planning`、`/persona`、`/add-dir` 暂未接通（会返回"暂不支持"提示）。 +- 以 `dsh --profile headless <提示>` 运行，`cwd` = 每用户会话目录（每用户工作区；模型生成的文件落在那里，可经 CDN 回传）。 +- 认证 / profile 为**机器级共享**（`~/.dsh`，与 grok 机器级登录同模型）：子进程 `HOME` 指向每用户会话目录，所以桥总是显式传 `DSH_HOME`。可用 `WECHATBRIDGE_DSH_HOME` 覆盖（如专用服务主目录；如需按用户隔离 `DSH_HOME`，需在该目录预置 profiles）。`DSH_BIN_PATH`、`DSH_PROFILE`、`DSH_TIMEOUT` 均可配置。 +- **状态：** 基于已发布的 dsh CLI 契约（headless bundle 源码）与测试用 fake CLI 实现；最终需由真实用户在 `dsh login` + headless 实跑后验收。 ### grok 后端说明 @@ -62,7 +70,8 @@ WeChatBridge 把微信机器人接到 agentic 编程 CLI（谷歌 agy / Antigrav - **agy** 在 `PATH` 中，或设 `AGY_BIN_PATH` - **和/或 grok** 在 `PATH` 中，或设 `GROK_BIN_PATH` - **和/或 codex** 在 `PATH` 中，或设 `CODEX_BIN_PATH` - - Antigravity 是谷歌终端 agentic 编程工具（Gemini CLI 官方继任）；Grok Build 是 xAI 同类产品；Codex 是 OpenAI 终端 agentic 编程工具 + - **和/或 dsh** 在 `PATH` 中，或设 `DSH_BIN_PATH`（DeepSeek Harness，需先 `dsh login`） + - Antigravity 是谷歌终端 agentic 编程工具（Gemini CLI 官方继任）；Grok Build 是 xAI 同类产品；Codex 是 OpenAI 终端 agentic 编程工具；dsh 是 DeepSeek Harness 的 CLI - 一个微信账号 + [ClawBot / iLink](https://ilinkai.weixin.qq.com) 机器人（首次扫码绑定） - Python 3.10+ @@ -133,10 +142,14 @@ curl -o ~/.config/wechatbridge/.env https://raw.githubusercontent.com/dorokuma/w | `AGY_BIN_PATH` | `agy` | agy 可执行文件路径 | | `GROK_BIN_PATH` | `grok` | grok 可执行文件路径 | | `CODEX_BIN_PATH` | `codex` | codex 可执行文件路径 | -| `WECHATBRIDGE_BACKEND` | `agy` | 全局默认后端（`agy` / `grok` / `codex`，可被 `/backend` 按用户覆盖） | +| `DSH_BIN_PATH` | `dsh` | dsh 可执行文件路径 | +| `DSH_PROFILE` | `headless` | dsh 单轮任务启动的 profile | +| `DSH_TIMEOUT` | `600` | dsh CLI 执行超时秒数 | +| `WECHATBRIDGE_DSH_HOME` | _空_ | 传给 dsh 子进程的显式 `DSH_HOME`（空 = 机器级 `~/.dsh`） | +| `WECHATBRIDGE_BACKEND` | `agy` | 全局默认后端（`agy` / `grok` / `codex` / `dsh`，可被 `/backend` 按用户覆盖） | | `WECHATBRIDGE_INSTANCE` | `default` | 实例名；state / 会话 / 二维码路径由它派生 | | `WECHATBRIDGE_ALLOWED_SENDERS` | _空_ | 允许使用的微信 ID，逗号分隔（空 = 全开） | -| `AGY_TIMEOUT` | `600` | CLI 执行超时秒数（三个后端共用） | +| `AGY_TIMEOUT` | `600` | CLI 执行超时秒数（agy / grok / codex 后端） | | `WECHATBRIDGE_MAX_OUTBOUND_BYTES` | `104857600` | 回传微信文件大小上限（100 MB） | | `WECHATBRIDGE_MAX_INBOUND_BYTES` | `20971520` | 入站图片/文件下载后上限（20 MB） | | `WECHATBRIDGE_MAX_CONCURRENT` | `4` | 全局同时处理上限；同用户串行且排队不占全局槽；满了回「忙」 | @@ -222,7 +235,7 @@ launchctl load ~/Library/LaunchAgents/com.wechatbridge.plist | 指令 | 作用 | |---|---| | `/help` | 按当前后端列出支持指令 | -| `/backend ` | 按微信用户切换 CLI 后端（真切换时清除该后端续聊状态——agy/grok 标记与 codex `thread_id`/resume——下次起新会话；历史文件可能仍在，靠保留策略清理） | +| `/backend ` | 按微信用户切换 CLI 后端（真切换时清除该后端续聊状态——agy/grok 标记与 codex `thread_id`/resume——下次起新会话；历史文件可能仍在，靠保留策略清理） | | `/clear` 或 `/new` | 丢掉续聊标记，下次 CLI 起新对话（不会立刻删除历史文件） | | `/model <名称>` | 设模型（各后端均对照实时列表校验：agy/grok 走 CLI `models`；codex 走 `codex debug models`，失败或空再试 `--bundled`；未知名或列表拉取失败则拒绝且不写 prefs；见 `/models`） | | `/models` | 列出可用模型——agy/grok/codex 均向 CLI 实时查询（codex：`debug models`；仅在实时列表拉不到时才回退内置参考说明） | @@ -254,8 +267,10 @@ launchctl load ~/Library/LaunchAgents/com.wechatbridge.plist ## 已知限制 -- 依赖 agy 和/或 grok 和/或 codex，本身不是独立 agent。 +- 依赖 agy 和/或 grok 和/或 codex 和/或 dsh，本身不是独立 agent。 - **codex** 后端尚未在真实 Codex 订阅/CLI 上实测，仅经源码研究、JSONL fixture 与 fake CLI 测试验证，暂视为社区测试，待真实用户确认。 +- **dsh** 后端为单轮模式（`headless` profile 每次都是新会话），且尚未在真实 `dsh login` + headless 上实测，目前基于已发布的 headless bundle 契约与测试用 fake CLI 验证。 +- dsh 的模型 / 强度 / 模式 / 人格 slash 指令尚未接通；`/model`、`/fast`、`/planning`、`/persona`、`/add-dir` 在 dsh 后端会返回"暂不支持"提示。 - 语音只靠微信转写；无本地 ASR；转写为空会提示改打字。 - 不收发视频；不回原生语音气泡（未做 silk 编码）。 - 一个进程绑一个微信号；多号多实例（`WECHATBRIDGE_INSTANCE`）。 diff --git a/deploy/wechatbridge.env.example b/deploy/wechatbridge.env.example index cc35985..9971ea1 100644 --- a/deploy/wechatbridge.env.example +++ b/deploy/wechatbridge.env.example @@ -27,7 +27,23 @@ # codex (OpenAI Codex) CLI binary path. If empty or "codex", assumes in PATH. # CODEX_BIN_PATH=codex -# Active CLI backend: "agy", "grok", or "codex". Can be switched per-user via /backend command. +# dsh (DeepSeek Harness) CLI binary path. If empty or "dsh", assumes in PATH. +# DSH_BIN_PATH=dsh + +# dsh profile booted for one-shot tasks. The headless profile answers a single +# task, prints the final assistant message, and exits (single-turn backend). +# DSH_PROFILE=headless + +# dsh execution timeout (seconds). Default: 600 (10 minutes). +# DSH_TIMEOUT=600 + +# Explicit DSH_HOME passed to the dsh child process. Empty = machine-wide +# default (~/.dsh): profiles and credentials are shared host-wide (same model +# as the grok backend's machine-wide login). Set this to a dedicated service +# home (or a per-user home with pre-seeded profiles) for full isolation. +# WECHATBRIDGE_DSH_HOME= + +# Active CLI backend: "agy", "grok", "codex", or "dsh". Can be switched per-user via /backend command. # WECHATBRIDGE_BACKEND=agy # --- Storage paths --- diff --git a/tests/fake_dsh.py b/tests/fake_dsh.py new file mode 100755 index 0000000..304b34c --- /dev/null +++ b/tests/fake_dsh.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python3 +"""Fake `dsh` CLI for integration tests of wechatbridge.dsh. + +Simulates ``dsh --profile headless `` — prints the final assistant +message on stdout and exits 0 only for a completed turn. Behaviour is +controlled by FAKE_DSH_MODE (read from the environment each invocation): + + ok (default) reply "first: ", exit 0. + artifact_link reply with a markdown file:/// link, create the file. + artifact_relative reply with a relative ./doc.md link, create the file. + internal_metadata reply with links to a real file AND a .dsh/internal file + (tests that run_dsh does not treat DSH_HOME internals as + workspace artifacts is out of scope here; the link is + still under cwd so send_artifacts_back decides). + empty no stdout, exit 0. + fail stderr "boom: something went wrong", exit 1. + dsh_error stderr "dsh: LLMError: model is currently overloaded", + exit 1. + not_logged_in stderr "dsh: AuthError: please log in first", exit 1. + timeout sleep far longer than the bridge timeout. + +If FAKE_DSH_LOG is set (path), every invocation appends one line so tests can +assert exactly how many times (and with what args) the CLI was launched. +""" + +import os +import sys +import time + + +def _reply(prompt): + return "first(%s)" % prompt + + +def main(): + args = sys.argv[1:] + mode = os.environ.get("FAKE_DSH_MODE", "ok") + + log_path = os.environ.get("FAKE_DSH_LOG") + if log_path: + try: + with open(log_path, "a", encoding="utf-8") as lf: + lf.write("invoked mode=%s args=%s\n" % (mode, " ".join(args))) + except OSError: + pass + + # dsh --profile headless — the task is the last positional + if "--profile" in args: + idx = args.index("--profile") + profile = args[idx + 1] if len(args) > idx + 1 else "" + task_args = args[idx + 2:] + else: + profile = "" + task_args = args + prompt = " ".join(task_args) + + if mode == "timeout": + time.sleep(60) + return 0 + if mode == "empty": + return 0 + if mode == "fail": + sys.stderr.write("boom: something went wrong\n") + sys.stderr.flush() + return 1 + if mode == "dsh_error": + sys.stderr.write("dsh: LLMError: model is currently overloaded\n") + sys.stderr.flush() + return 1 + if mode == "not_logged_in": + sys.stderr.write("dsh: AuthError: please log in first\n") + sys.stderr.flush() + return 1 + + if mode == "artifact_link": + path = os.path.join(os.getcwd(), "result.txt") + with open(path, "w", encoding="utf-8") as f: + f.write("generated content") + sys.stdout.write("[result.txt](file://%s)\n%s\n" % (path, _reply(prompt))) + sys.stdout.flush() + return 0 + + if mode == "artifact_relative": + path = os.path.join(os.getcwd(), "doc.md") + with open(path, "w", encoding="utf-8") as f: + f.write("# doc") + sys.stdout.write("see [doc.md](./doc.md) for details\n%s\n" % _reply(prompt)) + sys.stdout.flush() + return 0 + + if mode == "internal_metadata": + real = os.path.join(os.getcwd(), "report.pdf") + with open(real, "wb") as f: + f.write(b"%PDF-1.4") + internal = os.path.join(os.getcwd(), ".dsh", "internal", "meta.json") + os.makedirs(os.path.dirname(internal), exist_ok=True) + with open(internal, "w", encoding="utf-8") as f: + f.write("{}") + sys.stdout.write( + "[report.pdf](file://%s) and [meta](file://%s)\n%s\n" + % (real, internal, _reply(prompt)) + ) + sys.stdout.flush() + return 0 + + sys.stdout.write("%s\n" % _reply(prompt)) + sys.stdout.flush() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_dsh.py b/tests/test_dsh.py new file mode 100644 index 0000000..813b75a --- /dev/null +++ b/tests/test_dsh.py @@ -0,0 +1,379 @@ +"""Unit + integration tests for wechatbridge.dsh (DeepSeek Harness backend).""" + +from __future__ import annotations + +import os +import shutil +import sys +import tempfile +import unittest +from unittest import mock + +_HERE = os.path.dirname(os.path.abspath(__file__)) +_FAKE_DSH = os.path.join(_HERE, "fake_dsh.py") + + +def _rmtree(path): + try: + shutil.rmtree(path) + except OSError: + pass + + +def _sanitize(user_id): + from wechatbridge.runner_common import sanitize_user_id + return sanitize_user_id(user_id) + + +class _FakeProc: + """Minimal stand-in for an asyncio subprocess used by spawn-capture tests.""" + + def __init__(self, stdout="", stderr="", rc=0, pid=9999): + self._so = stdout.encode("utf-8") + self._se = stderr.encode("utf-8") + self.returncode = rc + self.pid = pid + + async def communicate(self): + return self._so, self._se + + +class TestBuildDshCommand(unittest.TestCase): + def setUp(self): + from wechatbridge.config import config + self._patchers = [] + p = mock.patch.object(config, "dsh_binary_path", "dsh") + p.start() + self._patchers.append(p) + p2 = mock.patch.object(config, "dsh_profile", "headless") + p2.start() + self._patchers.append(p2) + + def tearDown(self): + for p in self._patchers: + p.stop() + + def test_basic(self): + from wechatbridge.dsh import _build_dsh_command + self.assertEqual( + _build_dsh_command("hello"), + ["dsh", "--profile", "headless", "hello"], + ) + + def test_prompt_is_last_positional(self): + from wechatbridge.dsh import _build_dsh_command + cmd = _build_dsh_command("a b c") + self.assertEqual(cmd[-1], "a b c") + # headless 永远单轮：不得出现 resume / thread id + self.assertNotIn("resume", cmd) + + def test_custom_profile(self): + from wechatbridge.dsh import _build_dsh_command + from wechatbridge.config import config + with mock.patch.object(config, "dsh_profile", "custom"): + self.assertEqual( + _build_dsh_command("hi"), + ["dsh", "--profile", "custom", "hi"], + ) + + +class TestExtractArtifacts(unittest.TestCase): + def _extract(self, text, cwd=""): + from wechatbridge.dsh import extract_artifacts + return extract_artifacts(text, cwd=cwd) + + def test_empty(self): + self.assertEqual(self._extract(""), []) + self.assertEqual(self._extract("no links here"), []) + + def test_file_uri_link(self): + arts = self._extract("see [report.pdf](file:///tmp/x/report.pdf)") + self.assertEqual(arts, [("report.pdf", "/tmp/x/report.pdf")]) + + def test_bare_file_uri(self): + arts = self._extract("wrote file:///tmp/x/out.md") + self.assertEqual(arts, [("out.md", "/tmp/x/out.md")]) + + def test_relative_link_resolves_against_cwd(self): + arts = self._extract("see [doc.md](./doc.md)", cwd="/srv/ws") + self.assertEqual(arts, [("doc.md", "/srv/ws/doc.md")]) + + def test_parent_relative_link(self): + arts = self._extract("see [conf](../conf.yaml)", cwd="/srv/ws/sub") + self.assertEqual(arts, [("conf", "/srv/ws/conf.yaml")]) + + def test_absolute_link(self): + arts = self._extract("see [out](/tmp/out.txt)") + self.assertEqual(arts, [("out", "/tmp/out.txt")]) + + def test_http_and_bare_names_ignored(self): + arts = self._extract( + "see [site](https://example.com) and [tool](grep) and [x](file:///ok.txt)" + ) + self.assertEqual(arts, [("x", "/ok.txt")]) + + def test_dedup(self): + arts = self._extract( + "a [x](file:///tmp/a.txt) b [x](file:///tmp/a.txt)" + ) + self.assertEqual(len(arts), 1) + + def test_urlencoded_paths(self): + arts = self._extract("see [报 告](file:///tmp/my%20report.pdf)") + self.assertEqual(arts, [("报 告", "/tmp/my report.pdf")]) + + +class TestStripFileLinks(unittest.TestCase): + def _strip(self, text): + from wechatbridge.dsh import _strip_file_links + return _strip_file_links(text) + + def test_strips_targets_keeps_names(self): + out = self._strip("see [report.pdf](file:///srv/x/report.pdf) and [doc](./doc.md)") + self.assertEqual(out, "see [report.pdf] and [doc]") + + def test_leaves_plain_text(self): + self.assertEqual(self._strip("just text"), "just text") + + +class _DshIntegrationBase: + """Shared setup: temp session dir + fake host DSH_HOME with credentials.""" + + def _setUp(self): + self.td = tempfile.mkdtemp() + self.addCleanup(lambda: _rmtree(self.td)) + self.host_home = os.path.join(self.td, "host-dsh") + os.makedirs(self.host_home, exist_ok=True) + cred = os.path.join(self.host_home, ".credentials.yaml") + with open(cred, "w", encoding="utf-8") as f: + f.write("provider: deepseek\n") + + from wechatbridge.config import config + self._patchers = [] + p = mock.patch.object(config, "session_base_dir", os.path.join(self.td, "sessions")) + p.start(); self._patchers.append(p) + # 写一个 shim，subprocess 跑 shim 时 exec 到 python fake_dsh.py + self.shim = os.path.join(self.td, "dsh-shim.py") + with open(self.shim, "w", encoding="utf-8") as f: + f.write( + "#!/usr/bin/env python3\n" + "import sys, os\n" + f"sys.argv[0] = {_FAKE_DSH!r}\n" + "sys.argv.insert(0, sys.executable)\n" + "os.execv(sys.executable, sys.argv)\n" + ) + os.chmod(self.shim, 0o755) + p2 = mock.patch.object(config, "dsh_binary_path", self.shim) + p2.start(); self._patchers.append(p2) + p3 = mock.patch.object(config, "dsh_home", self.host_home) + p3.start(); self._patchers.append(p3) + p4 = mock.patch.object(config, "dsh_timeout", 30) + p4.start(); self._patchers.append(p4) + + def _tearDown(self): + for p in self._patchers: + p.stop() + + async def _run(self, prompt, user_id="u-dsh", mode="ok", timeout=None): + from wechatbridge import dsh as dsh_mod + with mock.patch.dict(os.environ, {"FAKE_DSH_MODE": mode}, clear=False): + return await dsh_mod.run_dsh(prompt, user_id, timeout=timeout) + + +class TestRunDshIntegration(unittest.IsolatedAsyncioTestCase, _DshIntegrationBase): + def setUp(self): + self._setUp() + + def tearDown(self): + self._tearDown() + + async def test_ok_reply_and_first_flag(self): + display, artifacts = await self._run("hello", mode="ok") + self.assertEqual(display, "first(hello)") + self.assertEqual(artifacts, []) + sd = os.path.join(config_session_dir(), _sanitize("u-dsh")) + # 首条成功会打 .initialized.dsh 标记 + self.assertTrue(os.path.exists(os.path.join(sd, ".initialized.dsh"))) + + async def test_second_call_does_not_resume(self): + log = os.path.join(self.td, "dsh.log") + with mock.patch.dict(os.environ, {"FAKE_DSH_LOG": log}, clear=False): + await self._run("one", mode="ok") + await self._run("two", mode="ok") + with open(log, encoding="utf-8") as f: + lines = f.read().strip().splitlines() + self.assertEqual(len(lines), 2) + for line in lines: + self.assertNotIn("resume", line) + self.assertIn("--profile headless", line) + + async def test_artifact_file_uri(self): + display, artifacts = await self._run("make pdf", mode="artifact_link") + self.assertEqual(len(artifacts), 1) + name, path = artifacts[0] + self.assertEqual(name, "result.txt") + self.assertTrue(os.path.isfile(path)) + sd = os.path.join(config_session_dir(), _sanitize("u-dsh")) + # macOS /var -> /private/var 是符号链接，两端都 realpath 再比前缀 + self.assertTrue(os.path.realpath(path).startswith(os.path.realpath(sd))) + # 显示文本里的 file:/// 链接目标被剥掉 + self.assertIn("[result.txt]", display) + self.assertNotIn("file://", display) + + async def test_artifact_relative_link(self): + display, artifacts = await self._run("doc", mode="artifact_relative") + self.assertEqual(len(artifacts), 1) + name, path = artifacts[0] + self.assertEqual(name, "doc.md") + sd = os.path.join(config_session_dir(), _sanitize("u-dsh")) + self.assertTrue( + os.path.realpath(path).startswith(os.path.realpath(sd)) + and path.endswith("doc.md") + ) + self.assertNotIn("./doc.md", display) + + async def test_empty_reply(self): + display, artifacts = await self._run("hi", mode="empty") + self.assertEqual(display, "（这次没有文字回复）") + self.assertEqual(artifacts, []) + + async def test_nonzero_exit_maps_to_error_bubble(self): + display, artifacts = await self._run("boom", mode="fail") + self.assertEqual(artifacts, []) + self.assertIn("❌", display) + # 首条失败不得打 initialized + sd = os.path.join(config_session_dir(), _sanitize("u-dsh")) + self.assertFalse(os.path.exists(os.path.join(sd, ".initialized.dsh"))) + + async def test_dsh_error_maps_to_generic_failure(self): + display, _ = await self._run("overloaded", mode="dsh_error") + self.assertIn("❌", display) + self.assertIn("执行失败", display) + + async def test_not_logged_in_maps_to_auth_bubble(self): + display, _ = await self._run("hi", mode="not_logged_in") + self.assertIn("未登录", display) + + async def test_missing_credentials_preflight(self): + # 删除宿主凭据 → 预检直接返回未登录，不拉起子进程 + cred = os.path.join(self.host_home, ".credentials.yaml") + os.remove(cred) + display, artifacts = await self._run("hi", mode="ok") + self.assertEqual(artifacts, []) + self.assertIn("未登录", display) + + async def test_timeout_returns_friendly_error(self): + display, _ = await self._run("slow", mode="timeout", timeout=0.3) + self.assertIn("超时", display) + + async def test_oversized_prompt_rejected(self): + display, artifacts = await self._run("x" * (130 * 1024), mode="ok") + self.assertEqual(artifacts, []) + self.assertIn("消息过长", display) + + +class TestDshSpawnEnv(unittest.IsolatedAsyncioTestCase, _DshIntegrationBase): + """DSH_HOME 显式传给子进程、HOME=session_dir、剥离 DSH_SESSION_* 变量。""" + + def setUp(self): + self._setUp() + + def tearDown(self): + self._tearDown() + + async def test_env_and_cwd(self): + captured = [] + + async def spawn(*args, **kwargs): + captured.append((list(args), kwargs)) + return _FakeProc("ok\n", "", 0) + + from wechatbridge import dsh as dsh_mod + with mock.patch.dict( + os.environ, + {"DSH_SESSION_ID": "leak", "DSH_SESSION_JSONL": "/leak/session.jsonl", "DSH_SHELL": "1"}, + clear=False, + ), mock.patch("asyncio.create_subprocess_exec", side_effect=spawn): + display, _ = await dsh_mod.run_dsh("hello", "u-env") + + self.assertEqual(display, "ok") + self.assertEqual(len(captured), 1) + argv, kwargs = captured[0] + self.assertEqual(argv[0], self.shim) + self.assertEqual(argv[1:], ["--profile", "headless", "hello"]) + sd = os.path.join(config_session_dir(), _sanitize("u-env")) + self.assertEqual(kwargs["cwd"], sd) + env = kwargs["env"] + self.assertEqual(env["DSH_HOME"], self.host_home) + self.assertEqual(env["HOME"], sd) + self.assertNotIn("DSH_SESSION_ID", env) + self.assertNotIn("DSH_SESSION_JSONL", env) + self.assertNotIn("DSH_SHELL", env) + self.assertEqual(env["PAGER"], "cat") + self.assertEqual(env["CI"], "true") + + +class TestDshSlashCommands(unittest.IsolatedAsyncioTestCase): + async def _handle(self, text, user_id="u-slash"): + from wechatbridge.dsh import handle_dsh_slash_command + return await handle_dsh_slash_command(text, user_id) + + async def test_help(self): + out = await self._handle("/help") + self.assertIn("dsh", out) + self.assertIn("/backend", out) + + async def test_clear_is_single_turn_notice(self): + out = await self._handle("/clear") + self.assertIn("单轮", out) + out2 = await self._handle("/new") + self.assertIn("单轮", out2) + + async def test_model_commands_not_supported(self): + for cmd in ("/model gemini-3", "/models", "/fast", "/planning", "/persona hi", "/add-dir /tmp", "/agents"): + out = await self._handle(cmd) + self.assertIn("不支持", out, cmd) + + async def test_dangerous_rejected(self): + for cmd in ("/exit", "/quit", "/logout"): + out = await self._handle(cmd) + self.assertIn("禁用", out) + + async def test_tui_panels_not_supported(self): + out = await self._handle("/config") + self.assertIn("微信端不支持", out) + + async def test_passthrough_returns_none(self): + self.assertIsNone(await self._handle("普通消息")) + self.assertIsNone(await self._handle("/whatever-custom")) + + +class TestDshBackendRegistration(unittest.TestCase): + def test_known_backends_includes_dsh(self): + from wechatbridge.runner_common import KNOWN_BACKENDS + self.assertIn("dsh", KNOWN_BACKENDS) + + def test_switch_backend_prefs_to_dsh(self): + from wechatbridge.runner_common import ( + KNOWN_BACKENDS, default_prefs, switch_backend_prefs, + ) + prefs = default_prefs() + prefs["backend"] = "agy" + old, new = switch_backend_prefs(prefs, "dsh") + self.assertEqual((old, new), ("agy", "dsh")) + self.assertEqual(prefs["backend"], "dsh") + self.assertIn("dsh", prefs["by_backend"]) + + def test_default_prefs_has_dsh_slot(self): + from wechatbridge.runner_common import default_prefs + prefs = default_prefs() + self.assertIn("dsh", prefs["by_backend"]) + + +def config_session_dir(): + from wechatbridge.config import config + return config.session_base_dir + + +if __name__ == "__main__": + unittest.main() diff --git a/wechatbridge/config.py b/wechatbridge/config.py index b2d9dfb..208900b 100644 --- a/wechatbridge/config.py +++ b/wechatbridge/config.py @@ -143,9 +143,9 @@ class AppConfig: # iLink base URL (no trailing slash) ilink_base_url: str = os.getenv("ILINK_BASE_URL", "https://ilinkai.weixin.qq.com") - # Active CLI backend: "agy", "grok", or "codex" (global default, can be overridden per-user via /backend) + # Active CLI backend: "agy", "grok", "codex", or "dsh" (global default, can be overridden per-user via /backend) backend: str = os.getenv("WECHATBRIDGE_BACKEND", "agy").lower() - if backend not in ("agy", "grok", "codex"): + if backend not in ("agy", "grok", "codex", "dsh"): logger.warning("Unknown backend %r, falling back to 'agy'", backend) backend = "agy" @@ -158,6 +158,21 @@ class AppConfig: # codex CLI binary path codex_binary_path: str = os.getenv("CODEX_BIN_PATH", "codex") # default assumes in PATH + # dsh (DeepSeek Harness) CLI binary path + dsh_binary_path: str = os.getenv("DSH_BIN_PATH", "dsh") # default assumes in PATH + + # dsh profile booted for one-shot tasks. The headless profile answers a + # single task, prints the final assistant message, and exits. + dsh_profile: str = os.getenv("DSH_PROFILE", "headless") + + # dsh execution timeout (seconds) + dsh_timeout: int = _env_int("DSH_TIMEOUT", 600) + + # Explicit DSH_HOME passed to the dsh child process. Empty = machine-wide + # default (~/.dsh): profiles and credentials are shared host-wide (same + # model as the grok backend's machine-wide login). + dsh_home: str = os.getenv("WECHATBRIDGE_DSH_HOME", "") + # Instance name (for multi-instance deployments) instance: str = _instance diff --git a/wechatbridge/dsh.py b/wechatbridge/dsh.py new file mode 100644 index 0000000..9af72a2 --- /dev/null +++ b/wechatbridge/dsh.py @@ -0,0 +1,346 @@ +""" +dsh (DeepSeek Harness CLI) runner with per-user workspace isolation. + +Boots the ``headless`` profile for one-shot tasks:: + + dsh --profile headless "" + +The headless bundle always creates a *fresh* session per invocation +(``session-``), prints the final assistant message to stdout, writes +``dsh: : `` to stderr on error, and exits 0 only when the turn +completed. This backend is therefore **single-turn**: every WeChat message +starts a new dsh session, so ``/clear`` / ``/new`` are accepted but are no-ops. + +Isolation mirrors the grok backend: the child runs with ``cwd`` = the per-user +session directory and ``HOME`` pointed there, while ``DSH_HOME`` stays +machine-wide (default ``~/.dsh``) so profiles and credentials are shared +host-wide like grok's machine-wide login. Override with +``WECHATBRIDGE_DSH_HOME`` to point ``DSH_HOME`` elsewhere (e.g. a dedicated +service home, or a per-user home when you pre-seed profiles). +""" + +import asyncio +import logging +import os +import re +import time +from urllib.parse import unquote + +from .config import config +from .runner_common import ( + clean_output, + ensure_session_dir, + format_error, + format_cli_error, + is_bridge_formatted_reply, + is_dangerous, + is_first_message, + mark_initialized, + sanitize_env, + terminate_process, + EMPTY_REPLY, +) + +logger = logging.getLogger("dsh_runner") + +# execve 单参数上限（Linux MAX_ARG_STRLEN = 128KB），留安全余量 +_MAX_ARG_BYTES = 120 * 1024 + +# 从子进程环境里剥掉的 dsh 会话相关变量：桥进程自身可能跑在某个 dsh 会话里 +# （本机登录态），不能把这些泄漏给子进程。 +_DSH_SESSION_ENV_KEYS = ("DSH_SESSION_ID", "DSH_SESSION_JSONL", "DSH_SHELL") + + +def _host_dsh_home() -> str: + """Machine-wide DeepSeek Harness home used by the dsh child process. + + Precedence: ``WECHATBRIDGE_DSH_HOME`` > ``WECHATBRIDGE_HOST_HOME``/``~``. + The child env sets ``HOME`` to the per-user session dir, so without an + explicit ``DSH_HOME`` dsh would resolve its home under the session dir and + find no profiles — we always pass the host home explicitly. + """ + if getattr(config, "dsh_home", ""): + return os.path.expanduser(config.dsh_home) + host_home = os.environ.get("WECHATBRIDGE_HOST_HOME") or os.path.expanduser("~") + return os.path.join(host_home, ".dsh") + + +def extract_artifacts(text: str, cwd: str = "") -> list[tuple[str, str]]: + """Extract (name, absolute_path) tuples of file references from dsh output. + + Recognizes (deduplicated, order-preserved): + - markdown links ``[name](file:///abs/path)`` (agy-compatible) + - markdown links ``[name](/abs/path)`` and ``[name](./rel/path)`` / + ``[name](../rel/path)`` — relative paths resolve against *cwd* + - bare ``file:///abs/path`` mentions + + Non-file URLs (``https://``, ``mailto:``, tool names) are ignored. + """ + if not text: + return [] + seen: set = set() + result: list[tuple[str, str]] = [] + + def _add(name: str, path: str) -> None: + if path.startswith("file://"): + # 只剥 "file://"（7 字符），保留 file:///abs 的第三个 "/"， + # 否则 /tmp/a.txt 会变成 tmp/a.txt 被当成相对路径丢弃。 + path = path[len("file://"):] + if not path.startswith("/"): + if cwd and (path.startswith("./") or path.startswith("../")): + path = os.path.normpath(os.path.join(cwd, path)) + else: + return # https://, mailto:, bare names, etc. + name = unquote(name.split("#")[0]) + path = unquote(path.split("#")[0]) + # 按路径去重（同一文件只回传一次；首个显示名优先） + if path not in seen: + seen.add(path) + result.append((name, path)) + + # [name](file:///path | /abs | ./rel | ../rel) + for m in re.finditer( + r"\[([^\]]+)\]\((file:///[^)\s]+|/(?:[^)\s]*)|\.\.?/[^)\s]+)\)", text + ): + _add(m.group(1), m.group(2)) + # bare file:///path + for m in re.finditer(r"file:///([^\s)\]}>]+)", text): + _add(os.path.basename(m.group(1)), "file:///" + m.group(1)) + + if result: + logger.debug("Extracted %d artifacts: %s", len(result), [n for n, _ in result[:3]]) + return result + + +def _strip_file_links(display: str) -> str: + """Remove file:/// and absolute/relative link targets from display text + so server paths never leak to WeChat users.""" + return re.sub( + r"\[([^\]]+)\]\((?:file:///|/|\.\.?/)[^)]+\)", + r"[\1]", + display, + ) + + +def _build_dsh_command(prompt: str) -> list: + """Build the dsh argv: ``dsh --profile ``. + + The headless profile joins its positionals into one task, so the prompt is + passed as a single positional. No continuation flag exists: headless + always creates a fresh session. + """ + return [config.dsh_binary_path, "--profile", config.dsh_profile, prompt] + + +# --------------------------------------------------------------------------- +# dsh CLI execution +# --------------------------------------------------------------------------- + +def clean_display(stdout_text: str) -> str: + """Clean CLI stdout for WeChat display and strip file link targets.""" + display = clean_output(stdout_text) or EMPTY_REPLY + return _strip_file_links(display) + + +async def run_dsh(prompt: str, user_id: str, timeout: int = None) -> tuple[str, list]: + """Execute the dsh headless profile for a single user message. + + - Runs with cwd = per-user session dir (per-user workspace isolation) + - Passes DSH_HOME explicitly (machine-wide, see _host_dsh_home) + - Single-turn: no session resume; every call is a fresh dsh session + - Extracts file artifacts from stdout, cleans ANSI/HTML from display text + - Kills the process group on timeout and returns a friendly message + + Returns: + tuple[str, list]: (cleaned_display_text, list_of_(name, abs_path)_artifacts) + """ + if timeout is None: + timeout = config.dsh_timeout + + if len(prompt.encode("utf-8", errors="replace")) > _MAX_ARG_BYTES: + logger.warning("Prompt too large for argv from user %s", user_id) + return format_error( + "消息过长", + f"这条消息太长了（超过 {_MAX_ARG_BYTES // 1024}KB），请精简或分段发送。", + ), [] + + t0 = time.time() + session_dir = ensure_session_dir(user_id) + + # Audit logging + logger.info("[AUDIT] user=%s prompt=%.200s", user_id, prompt) + if is_dangerous(prompt): + logger.warning("[AUDIT] dangerous keyword in prompt from user=%s", user_id) + + # Preflight: machine-wide credentials must exist (same spirit as agy's + # auth-token copy). Log the real path; never echo it to WeChat users. + cred_path = os.path.join(_host_dsh_home(), ".credentials.yaml") + if not os.path.exists(cred_path): + logger.warning("dsh credentials missing for user %s: %s", user_id, cred_path) + return format_error( + "未登录", + "助手尚未登录或凭证失效，请联系管理员处理。", + ), [] + + first = is_first_message(session_dir, backend="dsh") + cmd = _build_dsh_command(prompt) + logger.info( + "Running dsh for user %s (first=%s): %s", + user_id, first, " ".join(cmd[:3]) + " ...", + ) + + process = None + try: + env = sanitize_env(session_dir) + # sanitize_env 不会碰 DSH_*（非敏感名），必须显式接管： + # - DSH_HOME 指向机器级主目录（否则 HOME=session_dir 会让 dsh 解析到 + # 会话目录下、找不到 profile） + # - 剥离桥进程自身可能带有的 dsh 会话变量 + env["DSH_HOME"] = _host_dsh_home() + for k in _DSH_SESSION_ENV_KEYS: + env.pop(k, None) + env["PAGER"] = "cat" + env["CI"] = "true" + env["NONINTERACTIVE"] = "1" + env["PYTHONUNBUFFERED"] = "1" + + process = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=session_dir, + env=env, + preexec_fn=os.setsid if hasattr(os, "setsid") else None, + ) + + stdout_bytes, stderr_bytes = await asyncio.wait_for( + process.communicate(), + timeout=float(timeout), + ) + + stdout_text = stdout_bytes.decode("utf-8", errors="replace").strip() + stderr_text = stderr_bytes.decode("utf-8", errors="replace").strip() + + # Artifacts come from raw stdout (before clean_output) + artifacts = extract_artifacts(stdout_text, cwd=session_dir) + + display = clean_display(stdout_text) + + if process.returncode != 0: + logger.warning( + "dsh exited with code %s for user %s: %.200s", + process.returncode, + user_id, + stderr_text, + ) + raw = stderr_text.removeprefix("dsh: ").strip() or stdout_text or "process exited abnormally" + return format_cli_error(raw, backend="dsh"), [] + + # Success path only — never mark on ❌/🔔 error/throttle bubbles + if first and display != EMPTY_REPLY and not is_bridge_formatted_reply(display): + mark_initialized(session_dir, backend="dsh") + + elapsed = time.time() - t0 + logger.info( + "dsh done: user=%s elapsed=%.1fs artifacts=%d output=%d chars", + user_id, elapsed, len(artifacts), len(display), + ) + return display, artifacts + + except asyncio.TimeoutError: + logger.warning( + "dsh execution timed out after %ss for user %s", + timeout, + user_id, + ) + await terminate_process(process, graceful=True) + return format_error("处理超时", f"超过 {timeout} 秒未完成，已终止本次任务。"), [] + + except asyncio.CancelledError: + # 任务被取消（如重登录前排空）：必须杀掉子进程再传递取消 + await terminate_process(process, graceful=False) + raise + + except Exception as e: + logger.exception("Unexpected error running dsh: %s", e) + await terminate_process(process, graceful=False) + return format_error( + "执行出错", + "这次没处理好，请稍后再试。若一直失败，请联系管理员。", + ), [] + + +# --------------------------------------------------------------------------- +# Slash command support +# --------------------------------------------------------------------------- + +def _cmd_help() -> str: + """Build /help response listing dsh-supported slash commands.""" + lines = [ + "📋 **wechatbridge 支持指令 (dsh)** 📋", + "", + "**引擎说明**", + "- dsh 为**单轮模式**：每次提问都会开启全新会话", + "- `/backend ` — 切换助手引擎", + "", + "**对话控制**", + "- `/clear` 或 `/new` — dsh 单轮模式无需重置（指令已接受）", + "", + "**其他**", + "- `/help` — 显示本帮助", + "", + "提示：其他 `/` 指令会直接交给助手处理。", + ] + return "\n".join(lines) + + +async def handle_dsh_slash_command(text: str, user_id: str) -> str | None: + """Handle /-slash commands for the dsh backend. + + Classification (mirrors agy.py): + A — implemented here (help, clear/new, backend) + B — dangerous (exit, quit, logout) → rejected + C — TUI panels → not supported on WeChat + D — passthrough to dsh → returns None + + Returns: + str: reply message for A/B/C classes + None: for D class — the caller should pass the original text to run_dsh() + """ + parts = text.strip().split(maxsplit=1) + cmd = parts[0].lower() if parts else text.lower() + args = parts[1] if len(parts) > 1 else "" + + # --- B class: dangerous / rejected --- + B_CMDS = frozenset({"/exit", "/quit", "/logout"}) + if cmd in B_CMDS: + return "⛔ **该指令在微信端禁用** ⛔" + + # --- C class: TUI panels (not supported on WeChat) --- + C_CMDS = frozenset({ + "/config", "/settings", "/context", "/diff", "/artifact", "/tasks", + "/hooks", "/keybindings", "/permissions", "/statusline", + "/copy", "/open", "/rename", "/fork", "/branch", "/rewind", "/undo", + "/resume", "/switch", "/conversation", "/title", "/feedback", + "/usage", "/quota", "/credits", "/skills", + }) + if cmd in C_CMDS: + return f"⚠️ **微信端不支持** ⚠️\n\n`{cmd}`" + + # --- A class: implemented commands --- + if cmd == "/help": + return _cmd_help() + + if cmd in ("/clear", "/new"): + return "ℹ️ **dsh 为单轮模式** ℹ️\n\n每次提问都会开启全新会话，无需重置。" + + # v1: model / effort / mode / persona are not wired to dsh yet. + # /agent and /backend are meta-commands handled in main.py. + if cmd in ( + "/model", "/models", "/fast", "/planning", + "/add-dir", "/agents", "/persona", "/mcp", + ): + return "ℹ️ **该指令当前不支持 dsh 引擎** ℹ️\n\n请用 `/backend` 切换到 agy / grok / codex。" + + # --- D class: passthrough to dsh (return None so caller runs run_dsh) --- + return None diff --git a/wechatbridge/main.py b/wechatbridge/main.py index 80550ab..7a0e818 100644 --- a/wechatbridge/main.py +++ b/wechatbridge/main.py @@ -20,6 +20,7 @@ from .config import config, ensure_runtime_dirs from .ilink import ILinkClient from .runner_common import ( + KNOWN_BACKENDS, clean_session_media, clear_initialized, classify_upstream_failure, @@ -139,6 +140,9 @@ async def _run_llm(prompt: str, user_id: str) -> tuple[str, list]: elif backend == "codex": from .codex import run_codex return await run_codex(prompt, user_id) + elif backend == "dsh": + from .dsh import run_dsh + return await run_dsh(prompt, user_id) else: from .agy import run_agy return await run_agy(prompt, user_id) @@ -433,6 +437,9 @@ async def _handle_slash(client: ILinkClient, text: str, user_id: str, context_to elif backend == "codex": from .codex import handle_codex_slash_command return await handle_codex_slash_command(text, user_id) + elif backend == "dsh": + from .dsh import handle_dsh_slash_command + return await handle_dsh_slash_command(text, user_id) else: from .agy import handle_slash_command return await handle_slash_command(text, user_id) @@ -450,13 +457,18 @@ def _cmd_backend(args: str, user_id: str) -> str: prefs = load_prefs(user_id) current = prefs.get("backend", config.backend) model_label = format_model_label(prefs.get("model", "")) + backend_list = " / ".join(f"`{b}`" for b in KNOWN_BACKENDS) return ( f"📋 **当前助手引擎** 📋\n\n`{current}`\n" f"模型: `{model_label}`\n\n" - "用法: `/backend agy` 或 `/backend grok` 或 `/backend codex`" + f"用法: `/backend {backend_list}`" + ) + if name not in KNOWN_BACKENDS: + backend_list = " / ".join(f"`{b}`" for b in KNOWN_BACKENDS) + return ( + f"❌ **未知引擎** ❌\n\n支持: {backend_list}\n\n" + f"`/backend {'` 或 `/backend '.join(KNOWN_BACKENDS)}`" ) - if name not in ("agy", "grok", "codex"): - return "❌ **未知引擎** ❌\n\n支持: `agy` / `grok` / `codex`\n\n`/backend agy` 或 `/backend grok` 或 `/backend codex`" prefs = load_prefs(user_id) old, new = switch_backend_prefs(prefs, name) save_prefs(user_id, prefs) @@ -661,6 +673,11 @@ async def send_artifacts_back(client, from_user, context_token, artifacts) -> No # codex runs with cwd=session_dir; file_change paths may also land in # user-approved --add-dir roots, so allow those too. allowed_root = session_dir + elif backend == "dsh": + # dsh runs with cwd=session_dir (per-user workspace); artifacts are + # files the model created there. DSH_HOME sessions live outside the + # per-user tree (machine-wide ~/.dsh), so they never match anyway. + allowed_root = session_dir else: # agy writes to .gemini/antigravity-cli/scratch under session_dir allowed_root = os.path.join(session_dir, ".gemini", "antigravity-cli", "scratch") diff --git a/wechatbridge/runner_common.py b/wechatbridge/runner_common.py index 8884edf..9010cf4 100644 --- a/wechatbridge/runner_common.py +++ b/wechatbridge/runner_common.py @@ -681,7 +681,7 @@ def user_gap_remaining(self, user_id: str) -> float: # Per-user preference persistence (per-backend model/effort/mode memory) # --------------------------------------------------------------------------- -KNOWN_BACKENDS = ("agy", "grok", "codex") +KNOWN_BACKENDS = ("agy", "grok", "codex", "dsh") BACKEND_SCOPED_KEYS = ("model", "effort", "mode") @@ -1005,6 +1005,8 @@ def sanitize_env(session_dir: str) -> dict: os.path.join(".gemini", "antigravity-cli", "cache"), os.path.join(".grok", "logs"), os.path.join(".codex", "logs"), + os.path.join(".dsh", "logs"), + os.path.join(".dsh", "cache"), ) # Dialogue history — cleaned as *units* (never split SQLite sidecars / session trees). @@ -1015,6 +1017,7 @@ def sanitize_env(session_dir: str) -> dict: _HISTORY_KNOWLEDGE_REL = os.path.join(".gemini", "antigravity-cli", "knowledge") _HISTORY_GROK_SESSIONS_REL = os.path.join(".grok", "sessions") _HISTORY_CODEX_SESSIONS_REL = os.path.join(".codex", "sessions") +_HISTORY_DSH_SESSIONS_REL = os.path.join(".dsh", "sessions") _SESSION_HISTORY_REL_DIRS = ( _HISTORY_CONVERSATIONS_REL, @@ -1022,6 +1025,7 @@ def sanitize_env(session_dir: str) -> dict: _HISTORY_KNOWLEDGE_REL, _HISTORY_GROK_SESSIONS_REL, _HISTORY_CODEX_SESSIONS_REL, + _HISTORY_DSH_SESSIONS_REL, ) # SQLite sidecar suffixes that must share fate with the main ``*.db`` file. @@ -1401,6 +1405,11 @@ def _clean_user_history(user_dir: str, cutoff: float) -> int: removed += _clean_codex_sessions( os.path.join(user_dir, _HISTORY_CODEX_SESSIONS_REL), cutoff ) + # dsh flushes one session tree per run under .dsh/sessions/// + # — same bucket layout as grok, so the same unit-based cleaner applies. + removed += _clean_grok_sessions( + os.path.join(user_dir, _HISTORY_DSH_SESSIONS_REL), cutoff + ) return removed From 8dd9b5de55fd1f8601c57a644f275e80651724a0 Mon Sep 17 00:00:00 2001 From: dorokuma Date: Sat, 29 Aug 2026 15:44:50 +0800 Subject: [PATCH 2/2] fix: harden dsh backend before merge Security and correctness hardening for the dsh CLI backend: - Pass prompt as positional after -- to block option injection; fake_dsh stub now parses argv POSIX-style, with red/green tests - Strip bare file:/// links from display text; path extraction uses a whitelist so adjacent CJK text is not swallowed - Filter .dsh internal artifacts from send-back; clear .initialized.dsh on /clear - Machine-level session cleanup under DSH_HOME/sessions gated on explicit WECHATBRIDGE_DSH_HOME; unified strip+abspath normalization shared by runtime and cleanup; warn-once when unset - Replace @path mentions outside session_dir with [blocked-path]; token-based detection covers CJK, relative, symlink and adjacent-character bypass paths - Docs: -- in command examples, DSH_HOME semantics, attachment limits for dsh - Tests: 360 passing (baseline 325) --- CHANGELOG.md | 2 +- README.md | 7 +- README.zh-CN.md | 7 +- deploy/wechatbridge.env.example | 10 +- tests/fake_dsh.py | 45 +++++-- tests/test_dsh.py | 208 +++++++++++++++++++++++++++++++- tests/test_hardening.py | 177 +++++++++++++++++++++++++++ wechatbridge/config.py | 28 +++++ wechatbridge/dsh.py | 158 ++++++++++++++++++++---- wechatbridge/main.py | 7 +- wechatbridge/runner_common.py | 77 +++++++----- 11 files changed, 645 insertions(+), 81 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 89a6111..3b8d939 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ All notable changes follow [Keep a Changelog](https://keepachangelog.com/) and ### Added -- **dsh 后端**：新增 DeepSeek Harness CLI（`dsh`）作为第四个 CLI 后端（`/backend dsh`）。以 `dsh --profile headless ` 单轮运行，cwd 为每用户会话目录；从输出提取 `file:///` 与 markdown 文件链接作为可回传产物；`DSH_HOME` 机器级共享（可 `WECHATBRIDGE_DSH_HOME` 覆盖）；`/help`、`/clear`、`/new` 支持，模型/强度/人格类指令暂返回"暂不支持"。headless 每次新建会话，故为单轮模式。 +- **dsh 后端**：新增 DeepSeek Harness CLI（`dsh`）作为第四个 CLI 后端（`/backend dsh`）。以 `dsh --profile headless -- ` 单轮运行，cwd 为每用户会话目录；从输出提取 `file:///` 与 markdown 文件链接作为可回传产物（路径字符白名单避免误吞紧邻中文）；`DSH_HOME` 显式配置时启用专用主目录及自动会话保留清理，未配置时复用宿主 `~/.dsh` 且跳过自动清理；对会话目录外的 `@绝对路径` mention 予以屏蔽拦截（替换为 `[blocked-path]`）；`/help`、`/clear`、`/new` 支持，模型/强度/人格类指令暂返回"暂不支持"。headless 每次新建会话，故为单轮模式。 ### Tests diff --git a/README.md b/README.md index 126095c..a1cfb27 100644 --- a/README.md +++ b/README.md @@ -48,8 +48,9 @@ Per-user switch: `/backend agy`, `/backend grok`, `/backend codex`, or `/backend ### dsh backend notes - **Single-turn:** the `headless` profile always creates a fresh session per invocation (`session-`), so every WeChat message starts a new dsh session. `/clear` / `/new` are accepted but are no-ops, and `/model`, `/fast`, `/planning`, `/persona`, `/add-dir` are not wired yet (they return a short "not supported" notice). -- Runs `dsh --profile headless ` with `cwd` = the per-user session directory (per-user workspace; model-created files land there and can be sent back via CDN). -- Auth / profiles are **machine-wide** (`~/.dsh`, same model as grok's machine-wide login): the child's `HOME` points at the per-user session dir, so the bridge always passes `DSH_HOME` explicitly. Override with `WECHATBRIDGE_DSH_HOME` (e.g. a dedicated service home; for per-user `DSH_HOME` isolation, pre-seed profiles there). `DSH_BIN_PATH`, `DSH_PROFILE`, and `DSH_TIMEOUT` are configurable. +- Runs `dsh --profile headless -- ` with `cwd` = the per-user session directory (per-user workspace; model-created files land there and can be sent back via CDN). +- Image and file attachments are merged into the prompt text as `@/absolute/path` mentions only; whether the `headless` profile reads them has not been verified. The bridge blocks `@/absolute/path` mentions pointing outside the session directory (replaced with `[blocked-path]`). +- Auth / profiles are **machine-wide** (`~/.dsh`, same model as grok's machine-wide login): the child's `HOME` points at the per-user session dir, so the bridge always passes `DSH_HOME` explicitly. Set `WECHATBRIDGE_DSH_HOME` to configure a dedicated service home with automatic session retention cleanup; when unset, it reuses the host `~/.dsh` without automatic session cleanup (managed by the operator). `DSH_BIN_PATH`, `DSH_PROFILE`, and `DSH_TIMEOUT` are configurable. - Status: implemented from the published `dsh` CLI contract (headless bundle source) plus a fake CLI in the test suite; final acceptance depends on a real `dsh login` + headless run. ### Grok backend notes @@ -145,7 +146,7 @@ Key variables (all have defaults): | `DSH_BIN_PATH` | `dsh` | path to the dsh binary | | `DSH_PROFILE` | `headless` | dsh profile booted for one-shot tasks | | `DSH_TIMEOUT` | `600` | dsh CLI run timeout in seconds | -| `WECHATBRIDGE_DSH_HOME` | _empty_ | explicit `DSH_HOME` passed to the dsh child (empty = machine-wide `~/.dsh`) | +| `WECHATBRIDGE_DSH_HOME` | _empty_ | explicit `DSH_HOME` passed to the dsh child. Explicitly set = dedicated home + auto session cleanup; unset = reuse host `~/.dsh` without auto cleanup | | `WECHATBRIDGE_BACKEND` | `agy` | global default backend (`agy` / `grok` / `codex` / `dsh`; overridable per user via `/backend`) | | `WECHATBRIDGE_INSTANCE` | `default` | instance name; state / session / QR paths derive from it | | `WECHATBRIDGE_ALLOWED_SENDERS` | _empty_ | comma-separated WeChat IDs (empty = allow all) | diff --git a/README.zh-CN.md b/README.zh-CN.md index 5a0f734..17bbd8a 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -48,8 +48,9 @@ WeChatBridge 把微信机器人接到 agentic 编程 CLI（谷歌 agy / Antigrav ### dsh 后端说明 - **单轮模式：** `headless` profile 每次调用都新建会话（`session-`），所以每条微信消息都会开启全新 dsh 会话。`/clear` / `/new` 接受但无实际作用；`/model`、`/fast`、`/planning`、`/persona`、`/add-dir` 暂未接通（会返回"暂不支持"提示）。 -- 以 `dsh --profile headless <提示>` 运行，`cwd` = 每用户会话目录（每用户工作区；模型生成的文件落在那里，可经 CDN 回传）。 -- 认证 / profile 为**机器级共享**（`~/.dsh`，与 grok 机器级登录同模型）：子进程 `HOME` 指向每用户会话目录，所以桥总是显式传 `DSH_HOME`。可用 `WECHATBRIDGE_DSH_HOME` 覆盖（如专用服务主目录；如需按用户隔离 `DSH_HOME`，需在该目录预置 profiles）。`DSH_BIN_PATH`、`DSH_PROFILE`、`DSH_TIMEOUT` 均可配置。 +- 以 `dsh --profile headless -- <提示>` 运行，`cwd` = 每用户会话目录（每用户工作区；模型生成的文件落在那里，可经 CDN 回传）。 +- 图片/文件附件对 dsh 仅以 @绝对路径 文本并入 prompt，headless 是否读取未验证。桥对指向会话目录外的 @绝对路径 mention 进行了拦截屏蔽（替换为 `[blocked-path]`）。 +- 认证 / profile 为**机器级共享**（`~/.dsh`，与 grok 机器级登录同模型）：子进程 `HOME` 指向每用户会话目录，所以桥总是显式传 `DSH_HOME`。显式设置 `WECHATBRIDGE_DSH_HOME` 时使用专用主目录并开启自动会话保留清理；未设置时复用宿主 `~/.dsh`，不执行自动会话清理，由操作员自行管理。`DSH_BIN_PATH`、`DSH_PROFILE`、`DSH_TIMEOUT` 均可配置。 - **状态：** 基于已发布的 dsh CLI 契约（headless bundle 源码）与测试用 fake CLI 实现；最终需由真实用户在 `dsh login` + headless 实跑后验收。 ### grok 后端说明 @@ -145,7 +146,7 @@ curl -o ~/.config/wechatbridge/.env https://raw.githubusercontent.com/dorokuma/w | `DSH_BIN_PATH` | `dsh` | dsh 可执行文件路径 | | `DSH_PROFILE` | `headless` | dsh 单轮任务启动的 profile | | `DSH_TIMEOUT` | `600` | dsh CLI 执行超时秒数 | -| `WECHATBRIDGE_DSH_HOME` | _空_ | 传给 dsh 子进程的显式 `DSH_HOME`（空 = 机器级 `~/.dsh`） | +| `WECHATBRIDGE_DSH_HOME` | _空_ | 传给 dsh 子进程的显式 `DSH_HOME`。显式设置 = 专用目录+自动会话清理；未设 = 复用宿主 `~/.dsh` 且不自动清理 | | `WECHATBRIDGE_BACKEND` | `agy` | 全局默认后端（`agy` / `grok` / `codex` / `dsh`，可被 `/backend` 按用户覆盖） | | `WECHATBRIDGE_INSTANCE` | `default` | 实例名；state / 会话 / 二维码路径由它派生 | | `WECHATBRIDGE_ALLOWED_SENDERS` | _空_ | 允许使用的微信 ID，逗号分隔（空 = 全开） | diff --git a/deploy/wechatbridge.env.example b/deploy/wechatbridge.env.example index 9971ea1..0d50f12 100644 --- a/deploy/wechatbridge.env.example +++ b/deploy/wechatbridge.env.example @@ -37,10 +37,12 @@ # dsh execution timeout (seconds). Default: 600 (10 minutes). # DSH_TIMEOUT=600 -# Explicit DSH_HOME passed to the dsh child process. Empty = machine-wide -# default (~/.dsh): profiles and credentials are shared host-wide (same model -# as the grok backend's machine-wide login). Set this to a dedicated service -# home (or a per-user home with pre-seeded profiles) for full isolation. +# Explicit DSH_HOME passed to the dsh child process. +# When explicitly set: uses dedicated home directory and enables automatic +# session retention cleanup. +# When unset/empty: reuses machine-wide host default (~/.dsh), credentials and +# profiles are shared host-wide, and automatic session cleanup is disabled (left +# to the operator). # WECHATBRIDGE_DSH_HOME= # Active CLI backend: "agy", "grok", "codex", or "dsh". Can be switched per-user via /backend command. diff --git a/tests/fake_dsh.py b/tests/fake_dsh.py index 304b34c..da80a8b 100755 --- a/tests/fake_dsh.py +++ b/tests/fake_dsh.py @@ -36,24 +36,47 @@ def main(): args = sys.argv[1:] mode = os.environ.get("FAKE_DSH_MODE", "ok") + profile = "" + task_args = [] + i = 0 + in_options = True + + while i < len(args): + arg = args[i] + if in_options: + if arg == "--": + in_options = False + i += 1 + continue + if arg == "--profile": + if i + 1 < len(args): + profile = args[i + 1] + i += 2 + continue + else: + sys.stderr.write("dsh: error: option --profile requires an argument\n") + sys.stderr.flush() + return 1 + if arg.startswith("-"): + sys.stderr.write("dsh: error: unrecognized option: %s\n" % arg) + sys.stderr.flush() + return 1 + task_args.append(arg) + i += 1 + else: + task_args.append(arg) + i += 1 + + prompt = " ".join(task_args) + log_path = os.environ.get("FAKE_DSH_LOG") if log_path: try: with open(log_path, "a", encoding="utf-8") as lf: - lf.write("invoked mode=%s args=%s\n" % (mode, " ".join(args))) + lf.write("invoked mode=%s profile=%s task=%s args=%s\n" % (mode, profile, prompt, " ".join(args))) except OSError: pass - # dsh --profile headless — the task is the last positional - if "--profile" in args: - idx = args.index("--profile") - profile = args[idx + 1] if len(args) > idx + 1 else "" - task_args = args[idx + 2:] - else: - profile = "" - task_args = args - prompt = " ".join(task_args) - if mode == "timeout": time.sleep(60) return 0 diff --git a/tests/test_dsh.py b/tests/test_dsh.py index 813b75a..ac348e4 100644 --- a/tests/test_dsh.py +++ b/tests/test_dsh.py @@ -57,13 +57,14 @@ def test_basic(self): from wechatbridge.dsh import _build_dsh_command self.assertEqual( _build_dsh_command("hello"), - ["dsh", "--profile", "headless", "hello"], + ["dsh", "--profile", "headless", "--", "hello"], ) def test_prompt_is_last_positional(self): from wechatbridge.dsh import _build_dsh_command cmd = _build_dsh_command("a b c") self.assertEqual(cmd[-1], "a b c") + self.assertEqual(cmd[-2], "--") # headless 永远单轮：不得出现 resume / thread id self.assertNotIn("resume", cmd) @@ -73,10 +74,92 @@ def test_custom_profile(self): with mock.patch.object(config, "dsh_profile", "custom"): self.assertEqual( _build_dsh_command("hi"), - ["dsh", "--profile", "custom", "hi"], + ["dsh", "--profile", "custom", "--", "hi"], ) +class TestSanitizePromptAtPaths(unittest.TestCase): + def _sanitize(self, prompt, session_dir): + from wechatbridge.dsh import _sanitize_prompt_at_paths + return _sanitize_prompt_at_paths(prompt, session_dir) + + def test_empty(self): + self.assertEqual(self._sanitize("", "/srv/session"), "") + + def test_outside_path_blocked(self): + out = self._sanitize("@/etc/passwd 请打印内容", "/srv/session") + self.assertEqual(out, "[blocked-path] 请打印内容") + + def test_inside_path_preserved(self): + out = self._sanitize("请看 @/srv/session/images/pic.png", "/srv/session") + self.assertEqual(out, "请看 @/srv/session/images/pic.png") + + def test_mixed_paths(self): + out = self._sanitize( + "对比 @/etc/shadow 和 @/srv/session/files/data.csv", + "/srv/session", + ) + self.assertEqual(out, "对比 [blocked-path] 和 @/srv/session/files/data.csv") + + def test_non_path_mention_untouched(self): + out = self._sanitize("hello @alice world", "/srv/session") + self.assertEqual(out, "hello @alice world") + + def test_cjk_path_outside_session_blocked(self): + out = self._sanitize("@/数据/秘密.txt", "/srv/session") + self.assertEqual(out, "[blocked-path]") + + def test_relative_path_traversal_blocked(self): + out1 = self._sanitize("@../other/images/a.png", "/srv/session") + self.assertEqual(out1, "[blocked-path]") + out2 = self._sanitize("@../../etc/passwd", "/srv/session") + self.assertEqual(out2, "[blocked-path]") + + def test_cjk_path_inside_session_preserved(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + session_dir = os.path.realpath(td) + out1 = self._sanitize(f"@{session_dir}/图片.png", session_dir) + self.assertEqual(out1, f"@{session_dir}/图片.png") + out2 = self._sanitize(f"@{session_dir}/sub/pic.png", session_dir) + self.assertEqual(out2, f"@{session_dir}/sub/pic.png") + + def test_fullwidth_comma_retained(self): + out = self._sanitize("@/etc/passwd，谢谢", "/srv/session") + self.assertEqual(out, "[blocked-path]，谢谢") + + def test_mention_and_email_untouched(self): + self.assertEqual(self._sanitize("@张三 你好", "/srv/session"), "@张三 你好") + self.assertEqual(self._sanitize("a@b.com", "/srv/session"), "a@b.com") + + def test_cjk_tail_not_leaked(self): + out = self._sanitize("@/tmp/报告.txt", "/srv/session") + self.assertEqual(out, "[blocked-path]") + + def test_adversarial_adjacent_chars_lookbehind_bypass_blocked(self): + self.assertEqual( + self._sanitize("file@/etc/passwd", "/srv/session"), + "file[blocked-path]", + ) + self.assertEqual( + self._sanitize("user1@/etc/passwd", "/srv/session"), + "user1[blocked-path]", + ) + self.assertEqual( + self._sanitize("句子.@/etc/passwd", "/srv/session"), + "句子.[blocked-path]", + ) + + def test_real_session_dir_attachment_preserved(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + session_dir = os.path.realpath(td) + img_path = os.path.join(session_dir, "x.png") + out = self._sanitize(f"@{img_path}", session_dir) + self.assertEqual(out, f"@{img_path}") + + + class TestExtractArtifacts(unittest.TestCase): def _extract(self, text, cwd=""): from wechatbridge.dsh import extract_artifacts @@ -122,6 +205,35 @@ def test_urlencoded_paths(self): arts = self._extract("see [报 告](file:///tmp/my%20report.pdf)") self.assertEqual(arts, [("报 告", "/tmp/my report.pdf")]) + def test_internal_dsh_paths_filtered(self): + arts = self._extract( + "see [meta](file:///tmp/session/.dsh/internal/meta.json) and [doc](./doc.md)", + cwd="/tmp/session", + ) + self.assertEqual(arts, [("doc", "/tmp/session/doc.md")]) + + def test_internal_dsh_relative_paths_filtered(self): + arts = self._extract("see [meta](./.dsh/sessions/abc.json)", cwd="/tmp/session") + self.assertEqual(arts, []) + + def test_internal_dsh_bare_uri_filtered(self): + arts = self._extract("leak file:///tmp/session/.dsh/internal/meta.json") + self.assertEqual(arts, []) + + def test_bare_file_uri_with_adjacent_chinese(self): + arts = self._extract("见file:///tmp/a.pdf即可") + self.assertEqual(arts, [("a.pdf", "/tmp/a.pdf")]) + + def test_cjk_path_extraction(self): + arts = self._extract("file:///home/u/会话/报告.pdf") + self.assertEqual(arts, [("报告.pdf", "/home/u/会话/报告.pdf")]) + + def test_md_link_and_bare_uri_cjk_dedup(self): + arts = self._extract( + "[报告](file:///home/u/会话/报告.pdf) 以及 file:///home/u/会话/报告.pdf" + ) + self.assertEqual(arts, [("报告", "/home/u/会话/报告.pdf")]) + class TestStripFileLinks(unittest.TestCase): def _strip(self, text): @@ -132,9 +244,27 @@ def test_strips_targets_keeps_names(self): out = self._strip("see [report.pdf](file:///srv/x/report.pdf) and [doc](./doc.md)") self.assertEqual(out, "see [report.pdf] and [doc]") + def test_strips_bare_file_uri(self): + out = self._strip("已写入 file:///home/srv/x/report.pdf") + self.assertEqual(out, "已写入 ") + self.assertNotIn("file://", out) + self.assertNotIn("/home/srv/x", out) + + def test_strips_mixed_links(self): + out = self._strip("see [doc](file:///tmp/doc.txt) and bare file:///tmp/report.pdf here") + self.assertEqual(out, "see [doc] and bare here") + def test_leaves_plain_text(self): self.assertEqual(self._strip("just text"), "just text") + def test_strips_bare_file_uri_with_adjacent_chinese(self): + out = self._strip("见file:///tmp/a.pdf即可") + self.assertEqual(out, "见即可") + + def test_strips_bare_file_uri_with_cjk_path(self): + out = self._strip("file:///home/u/会话/报告.pdf") + self.assertEqual(out, "") + class _DshIntegrationBase: """Shared setup: temp session dir + fake host DSH_HOME with credentials.""" @@ -207,6 +337,27 @@ async def test_second_call_does_not_resume(self): self.assertNotIn("resume", line) self.assertIn("--profile headless", line) + async def test_prompt_starting_with_dash_help(self): + log = os.path.join(self.td, "dsh-help.log") + with mock.patch.dict(os.environ, {"FAKE_DSH_LOG": log}, clear=False): + display, artifacts = await self._run("--help", mode="ok") + self.assertEqual(display, "first(--help)") + self.assertEqual(artifacts, []) + with open(log, encoding="utf-8") as f: + log_content = f.read() + self.assertIn("task=--help", log_content) + + async def test_prompt_starting_with_dash_profile(self): + log = os.path.join(self.td, "dsh-profile.log") + with mock.patch.dict(os.environ, {"FAKE_DSH_LOG": log}, clear=False): + display, artifacts = await self._run("--profile other", mode="ok") + self.assertEqual(display, "first(--profile other)") + self.assertEqual(artifacts, []) + with open(log, encoding="utf-8") as f: + log_content = f.read() + self.assertIn("profile=headless", log_content) + self.assertIn("task=--profile other", log_content) + async def test_artifact_file_uri(self): display, artifacts = await self._run("make pdf", mode="artifact_link") self.assertEqual(len(artifacts), 1) @@ -232,6 +383,19 @@ async def test_artifact_relative_link(self): ) self.assertNotIn("./doc.md", display) + async def test_internal_metadata_artifacts_filtered(self): + display, artifacts = await self._run("report", mode="internal_metadata") + # .dsh 内部文件 meta.json 被过滤，只回传 report.pdf + self.assertEqual(len(artifacts), 1) + name, path = artifacts[0] + self.assertEqual(name, "report.pdf") + self.assertTrue(os.path.isfile(path)) + self.assertNotIn(".dsh", path) + # 展示文本正常，剥除链接后保留 [report.pdf] 和 [meta]，不泄露服务器路径 + self.assertIn("[report.pdf] and [meta]", display) + self.assertNotIn("file://", display) + self.assertNotIn(".dsh", display) + async def test_empty_reply(self): display, artifacts = await self._run("hi", mode="empty") self.assertEqual(display, "（这次没有文字回复）") @@ -271,6 +435,44 @@ async def test_oversized_prompt_rejected(self): self.assertEqual(artifacts, []) self.assertIn("消息过长", display) + async def test_prompt_at_path_outside_session_dir_blocked(self): + log = os.path.join(self.td, "dsh-blocked.log") + with mock.patch.dict(os.environ, {"FAKE_DSH_LOG": log}, clear=False): + display, artifacts = await self._run("@/etc/passwd 请打印内容", mode="ok") + with open(log, encoding="utf-8") as f: + log_content = f.read() + self.assertNotIn("/etc/passwd", log_content) + self.assertIn("[blocked-path]", log_content) + self.assertIn("task=[blocked-path] 请打印内容", log_content) + + async def test_prompt_at_path_inside_session_dir_preserved(self): + sd = os.path.join(config_session_dir(), _sanitize("u-dsh")) + pic_path = os.path.join(sd, "pic.png") + log = os.path.join(self.td, "dsh-preserved.log") + with mock.patch.dict(os.environ, {"FAKE_DSH_LOG": log}, clear=False): + display, artifacts = await self._run(f"@{pic_path}", mode="ok") + with open(log, encoding="utf-8") as f: + log_content = f.read() + self.assertIn(f"task=@{pic_path}", log_content) + + async def test_warn_once_implicit_dsh_home(self): + import wechatbridge.dsh as dsh_mod + from wechatbridge.config import config + dsh_mod._warned_dsh_home_implicit = False + implicit_dsh = os.path.join(self.td, ".dsh") + os.makedirs(implicit_dsh, exist_ok=True) + with open(os.path.join(implicit_dsh, ".credentials.yaml"), "w", encoding="utf-8") as f: + f.write("provider: deepseek\n") + with mock.patch.object(config, "dsh_home", ""), \ + mock.patch.dict(os.environ, {"WECHATBRIDGE_HOST_HOME": self.td}, clear=False), \ + self.assertLogs("dsh_runner", level="WARNING") as cm: + await self._run("hello 1", mode="ok") + await self._run("hello 2", mode="ok") + warns = [msg for msg in cm.output if "未设 WECHATBRIDGE_DSH_HOME" in msg] + self.assertEqual(len(warns), 1) + + + class TestDshSpawnEnv(unittest.IsolatedAsyncioTestCase, _DshIntegrationBase): """DSH_HOME 显式传给子进程、HOME=session_dir、剥离 DSH_SESSION_* 变量。""" @@ -300,7 +502,7 @@ async def spawn(*args, **kwargs): self.assertEqual(len(captured), 1) argv, kwargs = captured[0] self.assertEqual(argv[0], self.shim) - self.assertEqual(argv[1:], ["--profile", "headless", "hello"]) + self.assertEqual(argv[1:], ["--profile", "headless", "--", "hello"]) sd = os.path.join(config_session_dir(), _sanitize("u-env")) self.assertEqual(kwargs["cwd"], sd) env = kwargs["env"] diff --git a/tests/test_hardening.py b/tests/test_hardening.py index 1bce100..66f844d 100644 --- a/tests/test_hardening.py +++ b/tests/test_hardening.py @@ -910,6 +910,183 @@ def test_grok_branch_independent_of_codex(self): self.assertTrue(cleared.get("grok")) self.assertNotIn("codex", cleared) + def test_dsh_branch_clears_flag_when_no_history(self): + from wechatbridge.runner_common import _clear_initialized_if_no_history + + with tempfile.TemporaryDirectory() as user_dir: + dsh_flag = os.path.join(user_dir, ".initialized.dsh") + with open(dsh_flag, "w", encoding="utf-8") as f: + f.write("") + + cleared = _clear_initialized_if_no_history(user_dir) + + self.assertFalse(os.path.exists(dsh_flag)) + self.assertTrue(cleared.get("dsh")) + self.assertNotIn("grok", cleared) + self.assertNotIn("codex", cleared) + self.assertNotIn("agy", cleared) + + +class TestHostDshSessionCleanup(unittest.TestCase): + """Machine-wide $DSH_HOME/sessions/ cleanup during session data pruning.""" + + def test_expired_dsh_sessions_removed_and_fresh_kept(self): + from wechatbridge.runner_common import clean_session_data + from wechatbridge.config import config + + with tempfile.TemporaryDirectory() as td: + host_dsh = os.path.join(td, "dsh-home") + sessions_root = os.path.join(host_dsh, "sessions") + + # 1. Expired session under cwd1 (mtime = 40 days ago) + old_session = os.path.join(sessions_root, "cwd1", "session-old") + os.makedirs(old_session, exist_ok=True) + old_file = os.path.join(old_session, "chat.jsonl") + with open(old_file, "w", encoding="utf-8") as f: + f.write('{"role": "user"}\n') + old_mtime = time.time() - 40 * 86400 + os.utime(old_file, (old_mtime, old_mtime)) + os.utime(old_session, (old_mtime, old_mtime)) + + # 2. Fresh session under cwd1 (mtime = now) + fresh_session = os.path.join(sessions_root, "cwd1", "session-fresh") + os.makedirs(fresh_session, exist_ok=True) + fresh_file = os.path.join(fresh_session, "chat.jsonl") + with open(fresh_file, "w", encoding="utf-8") as f: + f.write('{"role": "user"}\n') + + # 3. Expired session under cwd2 (will become empty and cwd2 rmdir'd) + old_session2 = os.path.join(sessions_root, "cwd2", "session-old2") + os.makedirs(old_session2, exist_ok=True) + old_file2 = os.path.join(old_session2, "chat.jsonl") + with open(old_file2, "w", encoding="utf-8") as f: + f.write('{"role": "user"}\n') + os.utime(old_file2, (old_mtime, old_mtime)) + os.utime(old_session2, (old_mtime, old_mtime)) + + session_base = os.path.join(td, "user_sessions") + os.makedirs(session_base, exist_ok=True) + + with mock.patch.object(config, "dsh_home", host_dsh), \ + mock.patch.object(config, "session_base_dir", session_base), \ + mock.patch.object(config, "history_retention_days", 30): + removed = clean_session_data() + + self.assertGreaterEqual(removed, 2) + # Expired sessions removed + self.assertFalse(os.path.exists(old_session), "old session-old should be removed") + self.assertFalse(os.path.exists(old_session2), "old session-old2 should be removed") + # Fresh session preserved + self.assertTrue(os.path.exists(fresh_session), "fresh session should be preserved") + self.assertTrue(os.path.exists(fresh_file), "fresh file should be preserved") + # Empty bucket cwd2 should be pruned + self.assertFalse(os.path.exists(os.path.join(sessions_root, "cwd2")), "empty cwd2 bucket should be pruned") + # cwd1 still exists because it has fresh_session + self.assertTrue(os.path.exists(os.path.join(sessions_root, "cwd1")), "cwd1 bucket should still exist") + + def test_implicit_dsh_home_skips_cleanup(self): + from wechatbridge.runner_common import clean_session_data + from wechatbridge.config import config + + with tempfile.TemporaryDirectory() as td: + host_dsh = os.path.join(td, ".dsh") + sessions_root = os.path.join(host_dsh, "sessions") + + # Expired session under host ~/.dsh/sessions + old_session = os.path.join(sessions_root, "cwd1", "session-old") + os.makedirs(old_session, exist_ok=True) + old_file = os.path.join(old_session, "chat.jsonl") + with open(old_file, "w", encoding="utf-8") as f: + f.write('{"role": "user"}\n') + old_mtime = time.time() - 40 * 86400 + os.utime(old_file, (old_mtime, old_mtime)) + os.utime(old_session, (old_mtime, old_mtime)) + + session_base = os.path.join(td, "user_sessions") + os.makedirs(session_base, exist_ok=True) + + with mock.patch.object(config, "dsh_home", ""), \ + mock.patch.dict(os.environ, {"WECHATBRIDGE_HOST_HOME": td}, clear=False), \ + mock.patch.object(config, "session_base_dir", session_base), \ + mock.patch.object(config, "history_retention_days", 30): + removed = clean_session_data() + + self.assertEqual(removed, 0) + self.assertTrue( + os.path.exists(old_session), + "implicit host ~/.dsh session must NOT be removed", + ) + + def test_whitespace_dsh_home_not_explicit_and_skips_cleanup(self): + from wechatbridge.runner_common import clean_session_data + from wechatbridge.config import config, host_dsh_home, is_dsh_home_explicit + + with tempfile.TemporaryDirectory() as td: + host_dsh = os.path.join(td, ".dsh") + sessions_root = os.path.join(host_dsh, "sessions") + + old_session = os.path.join(sessions_root, "cwd1", "session-old") + os.makedirs(old_session, exist_ok=True) + old_file = os.path.join(old_session, "chat.jsonl") + with open(old_file, "w", encoding="utf-8") as f: + f.write('{"role": "user"}\n') + old_mtime = time.time() - 40 * 86400 + os.utime(old_file, (old_mtime, old_mtime)) + os.utime(old_session, (old_mtime, old_mtime)) + + session_base = os.path.join(td, "user_sessions") + os.makedirs(session_base, exist_ok=True) + + with mock.patch.object(config, "dsh_home", " "), \ + mock.patch.dict(os.environ, {"WECHATBRIDGE_HOST_HOME": td}, clear=False), \ + mock.patch.object(config, "session_base_dir", session_base), \ + mock.patch.object(config, "history_retention_days", 30): + self.assertFalse(is_dsh_home_explicit()) + self.assertEqual(host_dsh_home(), host_dsh) + removed = clean_session_data() + + self.assertEqual(removed, 0) + self.assertTrue(os.path.exists(old_session), "whitespace dsh_home must NOT remove host sessions") + + def test_relative_dsh_home_normalized_to_abspath_and_cleaned(self): + from wechatbridge.runner_common import clean_session_data + from wechatbridge.config import config, host_dsh_home, is_dsh_home_explicit + + with tempfile.TemporaryDirectory() as td: + rel_dir = os.path.join(".", "relative_dsh_home_test") + abs_dir = os.path.abspath(rel_dir) + sessions_root = os.path.join(abs_dir, "sessions") + self.addCleanup(lambda: shutil.rmtree(abs_dir, ignore_errors=True)) + + old_session = os.path.join(sessions_root, "cwd1", "session-old") + os.makedirs(old_session, exist_ok=True) + old_file = os.path.join(old_session, "chat.jsonl") + with open(old_file, "w", encoding="utf-8") as f: + f.write('{"role": "user"}\n') + old_mtime = time.time() - 40 * 86400 + os.utime(old_file, (old_mtime, old_mtime)) + os.utime(old_session, (old_mtime, old_mtime)) + + session_base = os.path.join(td, "user_sessions") + os.makedirs(session_base, exist_ok=True) + + with mock.patch.object(config, "dsh_home", rel_dir), \ + mock.patch.object(config, "session_base_dir", session_base), \ + mock.patch.object(config, "history_retention_days", 30): + self.assertTrue(is_dsh_home_explicit()) + self.assertEqual(host_dsh_home(), abs_dir) + removed = clean_session_data() + + self.assertGreaterEqual(removed, 1) + self.assertFalse(os.path.exists(old_session), "relative dsh_home session must be cleaned via abspath") + + def test_absolute_dsh_home_behavior_unchanged(self): + from wechatbridge.config import config, host_dsh_home, is_dsh_home_explicit + with mock.patch.object(config, "dsh_home", "/abs/path/to/dsh"): + self.assertTrue(is_dsh_home_explicit()) + self.assertEqual(host_dsh_home(), "/abs/path/to/dsh") + + class TestCleanCodexSessionsOSError(unittest.TestCase): """Regression: a single unreadable directory must not abort the whole codex diff --git a/wechatbridge/config.py b/wechatbridge/config.py index 208900b..77006b0 100644 --- a/wechatbridge/config.py +++ b/wechatbridge/config.py @@ -330,3 +330,31 @@ def ensure_runtime_dirs() -> None: os.chmod(path, 0o700) except OSError as e: logger.warning("Failed to ensure runtime dir %s: %s", path, e) + + +def _normalized_dsh_home() -> tuple[bool, str]: + """Single normalization helper for DSH_HOME. + + Returns: + tuple[bool, str]: (is_explicit, normalized_abs_path) + """ + raw = getattr(config, "dsh_home", "") + val = raw.strip() if raw else "" + if not val: + host_home = os.environ.get("WECHATBRIDGE_HOST_HOME") or os.path.expanduser("~") + return False, os.path.abspath(os.path.join(host_home, ".dsh")) + return True, os.path.abspath(os.path.expanduser(val)) + + +def host_dsh_home() -> str: + """Machine-wide DeepSeek Harness home used by dsh child process and session cleanup. + + Precedence: ``WECHATBRIDGE_DSH_HOME`` (config.dsh_home) > ``WECHATBRIDGE_HOST_HOME``/``~``. + """ + return _normalized_dsh_home()[1] + + +def is_dsh_home_explicit() -> bool: + """True when WECHATBRIDGE_DSH_HOME (config.dsh_home) was explicitly configured.""" + return _normalized_dsh_home()[0] + diff --git a/wechatbridge/dsh.py b/wechatbridge/dsh.py index 9af72a2..6060738 100644 --- a/wechatbridge/dsh.py +++ b/wechatbridge/dsh.py @@ -3,7 +3,7 @@ Boots the ``headless`` profile for one-shot tasks:: - dsh --profile headless "" + dsh --profile headless -- "" The headless bundle always creates a *fresh* session per invocation (``session-``), prints the final assistant message to stdout, writes @@ -11,12 +11,13 @@ completed. This backend is therefore **single-turn**: every WeChat message starts a new dsh session, so ``/clear`` / ``/new`` are accepted but are no-ops. -Isolation mirrors the grok backend: the child runs with ``cwd`` = the per-user -session directory and ``HOME`` pointed there, while ``DSH_HOME`` stays -machine-wide (default ``~/.dsh``) so profiles and credentials are shared -host-wide like grok's machine-wide login. Override with -``WECHATBRIDGE_DSH_HOME`` to point ``DSH_HOME`` elsewhere (e.g. a dedicated -service home, or a per-user home when you pre-seed profiles). +Workspace isolation: the child runs with ``cwd`` = the per-user session +directory (workspace for file artifacts) and ``HOME`` pointed there. +``DSH_HOME`` is pointed to a machine-wide host directory (default ``~/.dsh``, +or ``WECHATBRIDGE_DSH_HOME``) so credentials and profiles are shared host-wide. +As a trade-off, dsh session transcripts are written to the machine-wide +``$DSH_HOME/sessions/`` rather than per-user directories, and are expired by +the background cleanup runner based on age (cutoff). """ import asyncio @@ -26,7 +27,7 @@ import time from urllib.parse import unquote -from .config import config +from .config import config, host_dsh_home, is_dsh_home_explicit from .runner_common import ( clean_output, ensure_session_dir, @@ -36,6 +37,7 @@ is_dangerous, is_first_message, mark_initialized, + path_is_under, sanitize_env, terminate_process, EMPTY_REPLY, @@ -43,6 +45,8 @@ logger = logging.getLogger("dsh_runner") +_warned_dsh_home_implicit = False + # execve 单参数上限（Linux MAX_ARG_STRLEN = 128KB），留安全余量 _MAX_ARG_BYTES = 120 * 1024 @@ -59,10 +63,46 @@ def _host_dsh_home() -> str: explicit ``DSH_HOME`` dsh would resolve its home under the session dir and find no profiles — we always pass the host home explicitly. """ - if getattr(config, "dsh_home", ""): - return os.path.expanduser(config.dsh_home) - host_home = os.environ.get("WECHATBRIDGE_HOST_HOME") or os.path.expanduser("~") - return os.path.join(host_home, ".dsh") + return host_dsh_home() + + +_DSH_SKIP_DIR_NAMES = frozenset({ + ".dsh", + ".gemini", + ".codex", + ".grok", + ".git", + "__pycache__", + "node_modules", + "venv", + ".venv", + "cache", + ".cache", +}) + + +_BARE_FILE_URI_RE = re.compile(r"file:///[^\s)\]}>]+") +_TRAILING_PUNCT_CHARS = "。，！？；、：”’'\"）)]}>》」』〉…,.!?;:" + + +def _parse_bare_file_uri(raw: str) -> tuple[str, str]: + """Parse bare file:/// URI and separate trailing CJK commentary / punctuation. + + Returns: + tuple[str, str]: (clean_file_uri, trailing_kept_text) + """ + cleaned = raw.rstrip(_TRAILING_PUNCT_CHARS) + punct = raw[len(cleaned):] + last_slash = cleaned.rfind("/") + cjk_kept = "" + if last_slash != -1: + basename = cleaned[last_slash + 1:] + if re.search(r"[a-zA-Z0-9._~%+=:-][\u4e00-\u9fff]+$", basename): + cjk_m = re.search(r"[\u4e00-\u9fff]+$", basename) + if cjk_m: + cjk_kept = cjk_m.group(0) + cleaned = cleaned[:-len(cjk_kept)] + return cleaned, cjk_kept + punct def extract_artifacts(text: str, cwd: str = "") -> list[tuple[str, str]]: @@ -93,6 +133,10 @@ def _add(name: str, path: str) -> None: return # https://, mailto:, bare names, etc. name = unquote(name.split("#")[0]) path = unquote(path.split("#")[0]) + norm_path = os.path.normpath(path) + parts = norm_path.split(os.sep) + if any(part in _DSH_SKIP_DIR_NAMES for part in parts): + return # 按路径去重（同一文件只回传一次；首个显示名优先） if path not in seen: seen.add(path) @@ -103,9 +147,12 @@ def _add(name: str, path: str) -> None: r"\[([^\]]+)\]\((file:///[^)\s]+|/(?:[^)\s]*)|\.\.?/[^)\s]+)\)", text ): _add(m.group(1), m.group(2)) - # bare file:///path - for m in re.finditer(r"file:///([^\s)\]}>]+)", text): - _add(os.path.basename(m.group(1)), "file:///" + m.group(1)) + # bare file:///path with greedy match and trailing CJK/punctuation separation + for m in _BARE_FILE_URI_RE.finditer(text): + clean_uri, _ = _parse_bare_file_uri(m.group(0)) + path_part = clean_uri[len("file://"):] + if path_part.startswith("/"): + _add(os.path.basename(path_part), clean_uri) if result: logger.debug("Extracted %d artifacts: %s", len(result), [n for n, _ in result[:3]]) @@ -115,21 +162,81 @@ def _add(name: str, path: str) -> None: def _strip_file_links(display: str) -> str: """Remove file:/// and absolute/relative link targets from display text so server paths never leak to WeChat users.""" - return re.sub( + out = re.sub( r"\[([^\]]+)\]\((?:file:///|/|\.\.?/)[^)]+\)", r"[\1]", display, ) + def _replace_bare(m: re.Match) -> str: + _, kept = _parse_bare_file_uri(m.group(0)) + return kept + + return _BARE_FILE_URI_RE.sub(_replace_bare, out) + + +_AT_TOKEN_RE = re.compile(r"@([^\s@。，！？；、：”’'\"）)\]\}>》」』〉…]+)") +_AT_TRAILING_PUNCT = ".,!?;:)>]}'\"" + + +def _sanitize_prompt_at_paths(prompt: str, session_dir: str) -> str: + """Replace @ mentions pointing outside session_dir with [blocked-path]. + + Recognizes @ tokens (ASCII, CJK, relative ./ and ../, ~/ and file://). + Path candidates outside realpath(session_dir) are replaced with [blocked-path]. + Legitimate attachments under session_dir and non-path @mentions are preserved intact. + + Known limitation: + Paths containing spaces are truncated at the space by token splitting + and will be judged as blocked paths. + """ + if not prompt: + return "" + + def _replace(m: re.Match) -> str: + raw_token = m.group(1) + clean_token = raw_token.rstrip(_AT_TRAILING_PUNCT) + trailing_punct = raw_token[len(clean_token):] + + candidate = clean_token + # Determine whether token is a path candidate + is_candidate = ( + candidate.startswith(("/", "./", "../", "~/")) + or "://" in candidate + or candidate.startswith("file:") + ) + if not is_candidate: + return m.group(0) + + # Resolve path candidate + p = candidate + if p.startswith("file://"): + p = p[len("file://"):] + elif p.startswith("file:"): + p = p[len("file:"):] + + if p.startswith("~"): + p = os.path.expanduser(p) + + if not os.path.isabs(p): + target_path = os.path.normpath(os.path.join(session_dir, p)) + else: + target_path = p + + if path_is_under(target_path, session_dir): + return m.group(0) + return "[blocked-path]" + trailing_punct + + return _AT_TOKEN_RE.sub(_replace, prompt) def _build_dsh_command(prompt: str) -> list: - """Build the dsh argv: ``dsh --profile ``. + """Build the dsh argv: ``dsh --profile -- ``. The headless profile joins its positionals into one task, so the prompt is - passed as a single positional. No continuation flag exists: headless - always creates a fresh session. + passed as a single positional preceded by ``--`` to prevent option injection. + No continuation flag exists: headless always creates a fresh session. """ - return [config.dsh_binary_path, "--profile", config.dsh_profile, prompt] + return [config.dsh_binary_path, "--profile", config.dsh_profile, "--", prompt] # --------------------------------------------------------------------------- @@ -157,6 +264,11 @@ async def run_dsh(prompt: str, user_id: str, timeout: int = None) -> tuple[str, if timeout is None: timeout = config.dsh_timeout + global _warned_dsh_home_implicit + if not is_dsh_home_explicit() and not _warned_dsh_home_implicit: + logger.warning("未设 WECHATBRIDGE_DSH_HOME，复用宿主 ~/.dsh，宿主会话保留清理已禁用") + _warned_dsh_home_implicit = True + if len(prompt.encode("utf-8", errors="replace")) > _MAX_ARG_BYTES: logger.warning("Prompt too large for argv from user %s", user_id) return format_error( @@ -183,12 +295,16 @@ async def run_dsh(prompt: str, user_id: str, timeout: int = None) -> tuple[str, ), [] first = is_first_message(session_dir, backend="dsh") - cmd = _build_dsh_command(prompt) + + safe_prompt = _sanitize_prompt_at_paths(prompt, session_dir) + cmd = _build_dsh_command(safe_prompt) + logger.info( "Running dsh for user %s (first=%s): %s", user_id, first, " ".join(cmd[:3]) + " ...", ) + process = None try: env = sanitize_env(session_dir) diff --git a/wechatbridge/main.py b/wechatbridge/main.py index 7a0e818..98edcb5 100644 --- a/wechatbridge/main.py +++ b/wechatbridge/main.py @@ -675,8 +675,9 @@ async def send_artifacts_back(client, from_user, context_token, artifacts) -> No allowed_root = session_dir elif backend == "dsh": # dsh runs with cwd=session_dir (per-user workspace); artifacts are - # files the model created there. DSH_HOME sessions live outside the - # per-user tree (machine-wide ~/.dsh), so they never match anyway. + # files the model created there. Session transcripts live at machine-wide + # $DSH_HOME/sessions/ (outside per-user tree) and are cleaned by cutoff, + # so they never match anyway. allowed_root = session_dir else: # agy writes to .gemini/antigravity-cli/scratch under session_dir @@ -1277,7 +1278,7 @@ async def _safe_process_message(client: ILinkClient, msg: dict) -> None: # Entry point # --------------------------------------------------------------------------- def main(): - parser = argparse.ArgumentParser(prog="wechatbridge", description="Bridge WeChat messages to agy, Grok Build, or Codex CLIs — text/image/file/voice in, CLI replies and generated files back.") + parser = argparse.ArgumentParser(prog="wechatbridge", description="Bridge WeChat messages to agy, Grok Build, Codex, or dsh CLIs — text/image/file/voice in, CLI replies and generated files back.") parser.add_argument("--version", action="version", version=f"wechatbridge {__version__}") parser.parse_args() logger.info("wechatbridge v%s 启动 (backend=%s, instance=%s)", __version__, config.backend, config.instance) diff --git a/wechatbridge/runner_common.py b/wechatbridge/runner_common.py index 9010cf4..c1f50ee 100644 --- a/wechatbridge/runner_common.py +++ b/wechatbridge/runner_common.py @@ -17,7 +17,7 @@ import sys import time -from .config import config +from .config import config, host_dsh_home, is_dsh_home_explicit logger = logging.getLogger("wechatbridge.runner") @@ -1017,16 +1017,6 @@ def sanitize_env(session_dir: str) -> dict: _HISTORY_KNOWLEDGE_REL = os.path.join(".gemini", "antigravity-cli", "knowledge") _HISTORY_GROK_SESSIONS_REL = os.path.join(".grok", "sessions") _HISTORY_CODEX_SESSIONS_REL = os.path.join(".codex", "sessions") -_HISTORY_DSH_SESSIONS_REL = os.path.join(".dsh", "sessions") - -_SESSION_HISTORY_REL_DIRS = ( - _HISTORY_CONVERSATIONS_REL, - _HISTORY_BRAIN_REL, - _HISTORY_KNOWLEDGE_REL, - _HISTORY_GROK_SESSIONS_REL, - _HISTORY_CODEX_SESSIONS_REL, - _HISTORY_DSH_SESSIONS_REL, -) # SQLite sidecar suffixes that must share fate with the main ``*.db`` file. _SQLITE_SIDECAR_TAILS = ("-wal", "-shm", "-journal") @@ -1405,11 +1395,6 @@ def _clean_user_history(user_dir: str, cutoff: float) -> int: removed += _clean_codex_sessions( os.path.join(user_dir, _HISTORY_CODEX_SESSIONS_REL), cutoff ) - # dsh flushes one session tree per run under .dsh/sessions/// - # — same bucket layout as grok, so the same unit-based cleaner applies. - removed += _clean_grok_sessions( - os.path.join(user_dir, _HISTORY_DSH_SESSIONS_REL), cutoff - ) return removed @@ -1487,6 +1472,20 @@ def _clear_initialized_if_no_history(user_dir: str) -> dict: except OSError as e: logger.warning("Session cleanup: failed to clear %s: %s", flag, e) + # dsh: no per-user session history (transcripts live in host DSH_HOME/sessions); + # clear .initialized.dsh flag when cleaning session + flag = os.path.join(user_dir, ".initialized.dsh") + if os.path.exists(flag): + try: + os.remove(flag) + cleared["dsh"] = True + logger.info( + "Session cleanup: cleared .initialized.dsh for %s", + user_dir, + ) + except OSError as e: + logger.warning("Session cleanup: failed to clear %s: %s", flag, e) + # Legacy: clean shared .initialized if no history at all if not grok_has and not agy_has and not codex_has: flag = os.path.join(user_dir, ".initialized") @@ -1507,7 +1506,7 @@ def clean_session_data( retention_days: int | None = None, history_retention_days: int | None = None, ) -> int: - """Remove old session artifacts under each user directory. + """Remove old session artifacts under each user directory and host DSH_HOME. Two TTLs: - Temps (default ``session_retention_days``, often 7d): file-level mtime @@ -1517,6 +1516,8 @@ def clean_session_data( * agy ``conversations``: each ``*.db`` + ``*.db-wal/shm/journal`` together * agy ``brain`` / ``knowledge``: each top-level child tree/file * grok ``sessions``: each ``//`` tree; top-level indexes alone + * codex ``sessions``: date-bucketed ``rollout-*.jsonl`` files + * dsh ``sessions`` (machine-wide $DSH_HOME/sessions): each ``//`` tree Never splits a unit (prevents half-deleted SQLite DBs). Does **not** touch prefs, auth links, or CLI install trees. @@ -1531,25 +1532,37 @@ def clean_session_data( if history_retention_days is None: history_retention_days = config.history_retention_days base = config.session_base_dir - if not os.path.isdir(base): - return 0 now = time.time() temp_cutoff = now - max(int(retention_days), 0) * 86400 hist_cutoff = now - max(int(history_retention_days), 0) * 86400 removed = 0 - try: - for name in os.listdir(base): - user_dir = os.path.join(base, name) - if not os.path.isdir(user_dir): - continue - for rel in _SESSION_TEMP_REL_DIRS: - removed += _remove_old_files_under( - os.path.join(user_dir, rel), temp_cutoff - ) - removed += _clean_user_history(user_dir, hist_cutoff) - _clear_initialized_if_no_history(user_dir) - except OSError as e: - logger.error("Session data cleanup error: %s", e) + if os.path.isdir(base): + try: + for name in os.listdir(base): + user_dir = os.path.join(base, name) + if not os.path.isdir(user_dir): + continue + for rel in _SESSION_TEMP_REL_DIRS: + removed += _remove_old_files_under( + os.path.join(user_dir, rel), temp_cutoff + ) + removed += _clean_user_history(user_dir, hist_cutoff) + _clear_initialized_if_no_history(user_dir) + except OSError as e: + logger.error("Session data cleanup error: %s", e) + + # Machine-wide dsh session history cleanup (DSH_HOME/sessions///) + # Only runs when WECHATBRIDGE_DSH_HOME was explicitly configured to prevent + # wiping operator interactive sessions under default ~/.dsh + if is_dsh_home_explicit(): + try: + dsh_sessions = os.path.join(host_dsh_home(), "sessions") + if os.path.isdir(dsh_sessions): + removed += _clean_grok_sessions(dsh_sessions, hist_cutoff) + except Exception as e: + logger.error("DSH session cleanup error: %s", e) + + return removed