Files
AIGame/Tools/ws_probe_auth.py
T
udxandCursor b7a0bbcb89 Add hot-update boot pipeline, UI bundles/atlas assets, and design docs.
Ignore AIClient/Library and AIClient/HybridCLRData permanently; include HybridCLR/YooAsset project setup and UI tooling.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-21 16:02:58 +08:00

54 lines
1.8 KiB
Python

#!/usr/bin/env python3
import asyncio
import inspect
URL = "wss://game.xworld.ren/ws"
CASES = [
("pid_only", f"{URL}?pid=m2_probe", None),
("token_q", f"{URL}?pid=m2_probe&token=test", None),
("auth_q", f"{URL}?pid=m2_probe&auth=test", None),
("key_q", f"{URL}?pid=m2_probe&key=test", None),
("sig_q", f"{URL}?pid=m2_probe&sig=test", None),
("no_query", URL, None),
("bearer", f"{URL}?pid=m2_probe", [("Authorization", "Bearer test")]),
("x_token", f"{URL}?pid=m2_probe", [("X-Token", "test")]),
("x_api_key", f"{URL}?pid=m2_probe", [("X-Api-Key", "test")]),
]
async def try_one(websockets, name, url, headers):
kwargs = {"open_timeout": 8}
# websockets version differences
sig = inspect.signature(websockets.connect)
if headers:
if "additional_headers" in sig.parameters:
kwargs["additional_headers"] = headers
elif "extra_headers" in sig.parameters:
kwargs["extra_headers"] = headers
try:
async with websockets.connect(url, **kwargs) as ws:
print(f"OK {name}")
try:
msg = await asyncio.wait_for(ws.recv(), timeout=2)
print(f" recv {repr(msg)[:300]}")
except asyncio.TimeoutError:
print(" open but silent")
except Exception as e:
status = getattr(e, "status_code", None)
if status is None and hasattr(e, "response"):
status = getattr(e.response, "status_code", None)
print(f"FAIL {name}: {type(e).__name__} status={status} {e}")
async def main():
import websockets
print("websockets", websockets.__version__)
print("connect params", list(inspect.signature(websockets.connect).parameters))
for name, url, headers in CASES:
await try_one(websockets, name, url, headers)
if __name__ == "__main__":
asyncio.run(main())