udxandCursor
b7a0bbcb89
Ignore AIClient/Library and AIClient/HybridCLRData permanently; include HybridCLR/YooAsset project setup and UI tooling. Co-authored-by: Cursor <cursoragent@cursor.com>
82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
import asyncio
|
|
import json
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
BASE = "https://game.xworld.ren"
|
|
USER = f"m2_{int(time.time())}"
|
|
PASS = "m2_pass_123"
|
|
|
|
|
|
def http_json(method, path, payload=None):
|
|
data = None if payload is None else json.dumps(payload).encode()
|
|
req = urllib.request.Request(BASE + path, data=data, method=method)
|
|
if payload is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return resp.status, json.loads(resp.read().decode())
|
|
|
|
|
|
async def ws_try(url, headers=None):
|
|
import websockets
|
|
kwargs = {"open_timeout": 8}
|
|
if headers:
|
|
kwargs["extra_headers"] = headers
|
|
print("WS", url, headers)
|
|
async with websockets.connect(url, **kwargs) as ws:
|
|
print("WS OPEN")
|
|
try:
|
|
msg = await asyncio.wait_for(ws.recv(), timeout=3)
|
|
print("recv", repr(msg)[:500])
|
|
except asyncio.TimeoutError:
|
|
print("recv timeout")
|
|
# try hello
|
|
await ws.send(json.dumps({"op": "hello", "proto": 1}))
|
|
try:
|
|
msg = await asyncio.wait_for(ws.recv(), timeout=3)
|
|
print("after hello", repr(msg)[:500])
|
|
except asyncio.TimeoutError:
|
|
print("after hello timeout")
|
|
# ping style
|
|
for payload in (
|
|
{"op": "ping"},
|
|
{"type": "ping"},
|
|
"ping",
|
|
):
|
|
await ws.send(payload if isinstance(payload, str) else json.dumps(payload))
|
|
try:
|
|
msg = await asyncio.wait_for(ws.recv(), timeout=2)
|
|
print("after", payload, "=>", repr(msg)[:300])
|
|
except asyncio.TimeoutError:
|
|
print("after", payload, "=> timeout")
|
|
|
|
|
|
async def main():
|
|
print("register", USER)
|
|
st, reg = http_json("POST", "/register", {"username": USER, "password": PASS})
|
|
print("register", st, reg)
|
|
token = reg["token"]
|
|
pid = reg["pid"]
|
|
|
|
st, login = http_json("POST", "/login", {"username": USER, "password": PASS})
|
|
print("login", st, login)
|
|
|
|
cases = [
|
|
(f"wss://game.xworld.ren/ws?pid={pid}&token={token}", None),
|
|
(f"wss://game.xworld.ren/ws?pid={pid}", [("Authorization", f"Bearer {token}")]),
|
|
(f"wss://game.xworld.ren/ws?token={token}", None),
|
|
(f"wss://game.xworld.ren/ws?pid={pid}", [("X-Token", token)]),
|
|
]
|
|
for url, headers in cases:
|
|
try:
|
|
await ws_try(url, headers)
|
|
break
|
|
except Exception as e:
|
|
print("FAIL", type(e).__name__, e)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|