如何用Python模拟浏览器连接wss://socket.colonist.io WebSocket？

安全阿岳

2026-8-18

解决colonist.io WebSocket非浏览器端连接被立即关闭的问题

我需要实现对colonist.io的WebSocket交互自动化，目标连接地址为wss://socket.colonist.io。浏览器中执行以下JS代码可以正常接收消息：

ws = new WebSocket('wss://socket.colonist.io');
ws.onmessage = e => {
console.log(e);
}

但使用Python的websocket-client库（或Node.js、Python-Autobahn、Python-websockets）连接时，连接建立后立即被远程关闭。示例代码及追踪输出如下：

import websocket
def on_message(ws, data):
    print(f'received {data}')

websocket.enableTrace(True)
socket = websocket.WebSocketApp('wss://socket.colonist.io',
    header={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36'},
    on_message=on_message)
socket.run_forever(origin='https://colonist.io')
socket.close()

追踪输出：

--- request header ---
GET / HTTP/1.1
Upgrade: websocket
Host: socket.colonist.io
Origin: https://colonist.io
Sec-WebSocket-Key: EE3U0EDp36JGZBHWUN5q4Q==
Sec-WebSocket-Version: 13
Connection: Upgrade
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36


-----------------------
--- response header ---
HTTP/1.1 101 Switching Protocols
Server: nginx/1.18.0 (Ubuntu)
Date: Sat, 24 Sep 2022 17:33:32 GMT
Connection: upgrade
Upgrade: websocket
Sec-WebSocket-Accept: EwMJ+z82BuOBOSWONpuOhjNdVCQ=
-----------------------
websocket connected
Connection to remote host was lost. - goodbye

我怀疑服务端检测到非浏览器连接（已设置User-Agent和Origin）并关闭连接，希望找到无需Selenium的非浏览器端连接方案，以支持高并发连接。

问题分析

服务端并非仅通过User-Agent和Origin验证连接，大概率存在以下额外校验逻辑：

缺失关键请求头：浏览器WebSocket握手会携带Sec-WebSocket-Extensions（通常是permessage-deflate; client_max_window_bits）、会话Cookie等字段，Python库默认可能未包含。
未发送初始验证消息：浏览器连接成功后会立即发送特定的初始化消息（比如身份验证、版本协商的JSON数据），未发送该消息会被服务端主动断开。
会话一致性校验：服务端会验证WebSocket连接与浏览器访问的会话Cookie是否匹配，无有效Cookie会被拒绝。

解决方案

1. 捕获浏览器完整握手流程

用Chrome开发者工具（F12 → Network → 筛选WS）查看：

握手请求的全部请求头（重点看Cookie、Sec-WebSocket-Extensions）
连接成功后浏览器发送的第一条消息（在Frames标签中查看）

2. 补充完整请求头并携带会话Cookie

先通过HTTP请求获取colonist.io的会话Cookie，再在WebSocket请求中带上所有浏览器级别的请求头：

import asyncio
import websockets
import requests

# 获取会话Cookie
session = requests.Session()
session.get("https://colonist.io")
cookie_str = "; ".join([f"{k}={v}" for k, v in session.cookies.items()])

async def colonist_ws_connect():
    # 复制浏览器的请求头
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36",
        "Origin": "https://colonist.io",
        "Cookie": cookie_str,
        "Sec-WebSocket-Extensions": "permessage-deflate; client_max_window_bits"
    }

    async with websockets.connect(
        "wss://socket.colonist.io",
        extra_headers=headers,
        origin="https://colonist.io"
    ) as ws:
        # 发送浏览器捕获到的初始化消息（替换为实际内容）
        init_message = '{"type":"init","data":{"version":"0.1.234","client":"web"}}'
        await ws.send(init_message)
        
        # 持续接收消息
        async for msg in ws:
            print(f"Received: {msg}")

asyncio.run(colonist_ws_connect())

3. 适配服务端心跳机制

部分WebSocket服务会定期发送心跳包，若客户端未响应会断开连接。后续需要捕获浏览器的心跳交互逻辑，在Python代码中实现自动响应。

注意事项

务必确保请求头、初始化消息与浏览器完全一致，任何细微差异都可能触发服务端的反爬校验。
高并发场景下，建议使用异步库（如websockets）而非同步库，同时注意控制连接速率，避免触发服务端的限流机制。

内容的提问来源于stack exchange，提问作者Patrick