Cyrene/debug/cache/test_cdp_token_investigation.py

#!/usr/bin/env python3
"""调查 test-token-cyrene 的来源——检查 localStorage 和前端代码行为"""
import json, time, urllib.request
from websocket import create_connection

# 先获取真实 JWT
print("=== Step 0: 获取真实 JWT ===")
login_req = urllib.request.Request(
    'http://localhost:8080/api/v1/auth/login',
    data=json.dumps({"username": "yeij0942", "password": "Jiang1143218570"}).encode(),
    headers={"Content-Type": "application/json"},
    method='POST'
)
try:
    login_resp = urllib.request.urlopen(login_req, timeout=10)
    login_data = json.loads(login_resp.read())
    real_token = login_data.get('data', {}).get('access_token', '')
    print(f"  Real token (first 20 chars): {real_token[:20]}...")
    print(f"  Login success: {bool(real_token)}")
except Exception as e:
    print(f"  Login failed: {e}")
    real_token = ""

# 连接到 Chromium
print("\n=== Step 1: 创建新页面（about:blank，无旧 localStorage）===")
req = urllib.request.Request('http://127.0.0.1:9225/json/new?url=about:blank', method='PUT')
resp = urllib.request.urlopen(req, timeout=10)
page = json.loads(resp.read())
ws_url = page['webSocketDebuggerUrl']
print(f"  Page ID: {page.get('id')}")
print(f"  URL: {page.get('url')}")

ws = create_connection(ws_url, timeout=10)

def cdp(method, params=None, msg_id=1):
    ws.send(json.dumps({"id": msg_id, "method": method, "params": params or {}}))

def recv_all(timeout=3):
    ws.settimeout(timeout)
    msgs = []
    try:
        while True:
            msgs.append(ws.recv())
    except:
        pass
    return msgs

# 启用 domains
cdp("Page.enable", {}, 1)
cdp("Network.enable", {}, 2)
cdp("Runtime.enable", {}, 3)
time.sleep(0.5)
recv_all(0.5)

# === 检查 about:blank 的 localStorage ===
print("\n=== Step 2: 检查 about:blank localStorage（应为空）===")
cdp("Runtime.evaluate", {
    "expression": "JSON.stringify({token: localStorage.getItem('token'), allKeys: Object.keys(localStorage)})"
}, 10)
time.sleep(0.5)
msgs = recv_all(1)
for m in msgs:
    d = json.loads(m)
    if d.get('id') == 10:
        val = d.get('result', {}).get('result', {}).get('value', '')
        print(f"  about:blank localStorage: {val}")

# 如果有 token 就清除
cdp("Runtime.evaluate", {"expression": "localStorage.clear(); 'cleared'"}, 11)
time.sleep(0.5)
recv_all(0.5)

# === 导航到 5199 ===
print("\n=== Step 3: 导航到 localhost:5199 ===")
cdp("Page.navigate", {"url": "http://localhost:5199/"}, 20)
time.sleep(5)
all_msgs = recv_all(3)

# 分析网络请求
print("\n=== Step 4: 网络请求分析 ===")
for m in all_msgs:
    d = json.loads(m)
    method = d.get('method', '')
    params = d.get('params', {})
    if method == 'Network.requestWillBeSent':
        req = params.get('request', {})
        headers = req.get('headers', {})
        auth = headers.get('Authorization', '') or headers.get('authorization', '')
        url_short = req.get('url', '')[:100]
        if auth:
            print(f"  AUTH: {req.get('method','?')} {url_short}")
            print(f"        Authorization: {auth[:80]}")
        elif '/api/' in url_short:
            print(f"  NO_AUTH: {req.get('method','?')} {url_short}")
    elif method == 'Runtime.consoleAPICalled':
        args = params.get('args', [])
        texts = [str(a.get('value', '') or a.get('description', ''))[:150] for a in args]
        msg_type = params.get('type', 'log')
        print(f"  [CONSOLE:{msg_type}] {' '.join(texts)}")
    elif method == 'Runtime.exceptionThrown':
        exc = params.get('exceptionDetails', {})
        print(f"  [EXCEPTION] {exc.get('text', '')} at line {exc.get('lineNumber', '')}")

# === 检查 localStorage ===
print("\n=== Step 5: 检查导航后 localStorage ===")
cdp("Runtime.evaluate", {
    "expression": "JSON.stringify({token: localStorage.getItem('token'), refresh_token: localStorage.getItem('refresh_token'), user_id: localStorage.getItem('user_id'), allKeys: Object.keys(localStorage)})"
}, 30)
time.sleep(0.5)
msgs = recv_all(2)
for m in msgs:
    d = json.loads(m)
    if d.get('id') == 30:
        val = d.get('result', {}).get('result', {}).get('value', '')
        print(f"  localStorage: {val}")

# === 如果 token 是 test-token-cyrene，手动覆盖为真实 token 并测试 ===
print("\n=== Step 6: 注入真实 token ===")
if real_token:
    cdp("Runtime.evaluate", {
        "expression": f"localStorage.setItem('token', '{real_token}'); 'done'"
    }, 40)
    time.sleep(0.5)
    recv_all(1)
    
    # 重新加载
    cdp("Page.reload", {}, 41)
    time.sleep(4)
    msgs = recv_all(3)
    
    # 查看请求
    print("  重新加载后的网络请求:")
    for m in msgs:
        d = json.loads(m)
        method = d.get('method', '')
        params = d.get('params', {})
        if method == 'Network.requestWillBeSent':
            req = params.get('request', {})
            headers = req.get('headers', {})
            auth = headers.get('Authorization', '') or headers.get('authorization', '')
            url_short = req.get('url', '')[:100]
            if auth or '/api/' in url_short:
                if auth:
                    print(f"    AUTH: {req.get('method','?')} {url_short} -> {auth[:80]}")
                else:
                    print(f"    NO_AUTH: {req.get('method','?')} {url_short}")
        elif method == 'Runtime.consoleAPICalled':
            args = params.get('args', [])
            texts = [str(a.get('value', '') or a.get('description', ''))[:150] for a in args]
            print(f"    [CONSOLE:{params.get('type','log')}] {' '.join(texts)}")

ws.close()
print("\nDone.")