chuanqi/cq2_sdk/scripts/test_refresh_extend.py

"""Experiment: does getByRefresh extend the token's validity?

Method:
  1. Decode current JWT payload (iat/exp/refresh_exp) BEFORE the call
  2. Call POST user-api.996hezi.com/v1/UserInfo/getByRefresh with the token
  3. Dump the FULL raw response (every key — last time we only looked at nick/id)
  4. Decode JWT AFTER — compare; look for ANY new token string anywhere
     in the response (data.token / data.userinfo.token / set-cookie header etc.)
"""
import asyncio
import base64
import json
import sys
from datetime import datetime, timezone
from pathlib import Path

_HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(_HERE.parent / "src"))

from cq2_sdk.env_loader import load_env
load_env(_HERE.parent / ".env")

from cq2_sdk.login import Box996Login, ManualGeetestSolver
from cq2_sdk.store import UserStore, connection

def decode_jwt(token: str) -> dict:
    parts = token.split(".")
    pad = parts[1] + "=" * (-len(parts[1]) % 4)
    return json.loads(base64.urlsafe_b64decode(pad))


def show_life(payload: dict, label: str) -> None:
    def ts(k):
        v = payload.get(k)
        return f"{datetime.fromtimestamp(v, tz=timezone.utc).isoformat()[:16]}" if v else "-"
    print(f"  [{label}] iat={ts('iat')}  exp={ts('exp')}  refresh_exp={ts('refresh_exp')}")


def hunt_tokens(obj, path="") -> list:
    """Recursively find any long eyJ... strings anywhere in the response."""
    found = []
    if isinstance(obj, dict):
        for k, v in obj.items():
            found += hunt_tokens(v, f"{path}.{k}")
    elif isinstance(obj, list):
        for i, v in enumerate(obj):
            found += hunt_tokens(v, f"{path}[{i}]")
    elif isinstance(obj, str) and obj.startswith("eyJ") and len(obj) > 100:
        found.append((path, obj))
    return found


async def main() -> int:
    with connection() as conn:
        u = UserStore(conn).get_by_username("longwu988")
    old_token = u.token

    print("=== 1) JWT BEFORE ===")
    p_before = decode_jwt(old_token)
    show_life(p_before, "before")

    print("\n=== 2) call getByRefresh ===")
    box = Box996Login(geetest=ManualGeetestSolver(), device_id="00:15:5d:0a:04:00")
    box.token = old_token
    r = await box._request(
        "POST", "https://user-api.996hezi.com", "/v1/UserInfo/getByRefresh"
    )
    print(f"  code={r.get('code')}  msg={r.get('msg', '')!r}")

    print("\n=== 3) FULL raw response (every key) ===")
    print(json.dumps(r, ensure_ascii=False, indent=2)[:3000])

    print("\n=== 4) token hunt: any NEW eyJ... anywhere in response? ===")
    hits = hunt_tokens(r)
    if not hits:
        print("  (none found)")
    new_token = None
    for path, tok in hits:
        is_new = tok != old_token
        print(f"  {path}: len={len(tok)}  NEW={is_new}")
        if is_new and ".eyJ" in tok:
            new_token = tok
            try:
                show_life(decode_jwt(tok), "candidate-new")
            except Exception as e:
                print(f"    (decode failed: {e})")

    print("\n=== verdict ===")
    if new_token:
        p_after = decode_jwt(new_token)
        show_life(p_after, "after")
        gained = p_after.get("exp", 0) - p_before.get("exp", 0)
        print(f"  NEW TOKEN issued! exp extended by {gained/86400:.1f} days")
        with connection() as conn:
            UserStore(conn).update_token(user_id=u.id, token=new_token)
        print("  → saved to MySQL")
    else:
        exp_left = (p_before.get("exp", 0) - 1789700000) / 86400  # vs ~now
        print("  getByRefresh does NOT issue a new token.")
        print(f"  current token still valid until exp (above); plan: re-login via A2 before expiry.")
    return 0


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