小鱼开发
154 lines
5.0 KiB
Python
154 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
验证双 Token + 单设备踢人功能
|
|
==============================
|
|
|
|
用法：
|
|
cd python-api
|
|
python -m scripts.test_auth --mobile 13800138000
|
|
|
|
流程：
|
|
1. 发送验证码
|
|
2. 设备 A 登录（获取 Token A）
|
|
3. 用 Token A 调用 /me 验证 Access Token 有效
|
|
4. 用 Token A 的 refresh_token 刷新（验证 Token 轮换）
|
|
5. 设备 B 登录（同一手机号，获取 Token B）
|
|
6. 验证 Token A 已失效（被踢）
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
import time
|
|
|
|
import httpx
|
|
|
|
# 测试环境 API 地址
|
|
BASE_URL = "https://dev.tapi.meijiaka.cn/api/v1"
|
|
|
|
|
|
async def send_code(mobile: str) -> None:
|
|
"""发送验证码（开发环境验证码会打印到后端日志）"""
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.post(f"{BASE_URL}/auth/send-code", json={"mobile": mobile})
|
|
print(f"[1] 发送验证码: {r.status_code} {r.text}")
|
|
r.raise_for_status()
|
|
|
|
|
|
async def login(mobile: str, code: str, device_id: str) -> dict:
|
|
"""登录，返回 {access_token, refresh_token, user}"""
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.post(
|
|
f"{BASE_URL}/auth/login",
|
|
json={
|
|
"mobile": mobile,
|
|
"code": code,
|
|
"device_id": device_id,
|
|
"device_name": f"测试设备-{device_id[-4:]}",
|
|
"os_info": "test-script",
|
|
"app_version": "0.1.0",
|
|
},
|
|
)
|
|
print(f"[2] 登录 ({device_id}): {r.status_code}")
|
|
r.raise_for_status()
|
|
return r.json()["data"]
|
|
|
|
|
|
async def call_me(access_token: str) -> dict:
|
|
"""用 Access Token 调用 /me"""
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.get(
|
|
f"{BASE_URL}/auth/me",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
print(f"[3] /me: {r.status_code} {r.text[:200]}")
|
|
r.raise_for_status()
|
|
return r.json()["data"]
|
|
|
|
|
|
async def refresh_token(refresh_token: str) -> dict:
|
|
"""用 Refresh Token 换取新的 Token 对"""
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.post(
|
|
f"{BASE_URL}/auth/refresh",
|
|
json={"refresh_token": refresh_token},
|
|
)
|
|
print(f"[4] 刷新 Token: {r.status_code}")
|
|
r.raise_for_status()
|
|
return r.json()["data"]
|
|
|
|
|
|
async def verify_kicked(access_token: str) -> bool:
|
|
"""验证旧 Access Token 是否已失效（被踢）"""
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.get(
|
|
f"{BASE_URL}/auth/me",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
print(f"[6] 旧 Token 调用 /me: {r.status_code} {r.text[:200]}")
|
|
return r.status_code == 401
|
|
|
|
|
|
async def main():
|
|
parser = argparse.ArgumentParser(description="验证双 Token + 踢人功能")
|
|
parser.add_argument("--mobile", required=True, help="手机号")
|
|
parser.add_argument("--code", default="123456", help="验证码（默认 123456，需与日志一致）")
|
|
args = parser.parse_args()
|
|
|
|
mobile = args.mobile
|
|
code = args.code
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"测试手机号: {mobile}")
|
|
print(f"验证码: {code}")
|
|
print(f"{'='*60}\n")
|
|
|
|
# 1. 发送验证码
|
|
await send_code(mobile)
|
|
print("⚠️ 请确认后端日志中的验证码与 --code 一致\n")
|
|
|
|
# 2. 设备 A 登录
|
|
device_a = "device-a-test"
|
|
token_a = await login(mobile, code, device_a)
|
|
print(f" Access Token A: {token_a['access_token'][:40]}...")
|
|
print(f" Refresh Token A: {token_a['refresh_token'][:40]}...")
|
|
print()
|
|
|
|
# 3. 用 Token A 调用 /me
|
|
me = await call_me(token_a["access_token"])
|
|
print(f" 用户信息: {me['nickname']} ({me['mobile']})\n")
|
|
|
|
# 4. 用 Refresh Token A 刷新（验证 Token 轮换）
|
|
token_a2 = await refresh_token(token_a["refresh_token"])
|
|
print(f" New Access Token: {token_a2['access_token'][:40]}...")
|
|
print(f" New Refresh Token: {token_a2['refresh_token'][:40]}...")
|
|
print()
|
|
|
|
# 5. 设备 B 登录（同一手机号，触发踢人）
|
|
device_b = "device-b-test"
|
|
token_b = await login(mobile, code, device_b)
|
|
print(f" Access Token B: {token_b['access_token'][:40]}...")
|
|
print()
|
|
|
|
# 6. 验证 Token A（旧的）已失效
|
|
is_kicked = await verify_kicked(token_a["access_token"])
|
|
if is_kicked:
|
|
print("✅ 旧 Token A 已失效（被踢）—— 单设备登录生效\n")
|
|
else:
|
|
print("❌ 旧 Token A 仍然有效 —— 踢人逻辑有问题\n")
|
|
|
|
# 7. 验证 Token A2（刷新后的）也应该失效（因为设备 B 登录会覆盖设备记录）
|
|
is_kicked2 = await verify_kicked(token_a2["access_token"])
|
|
if is_kicked2:
|
|
print("✅ 刷新后的 Token A2 也已失效\n")
|
|
else:
|
|
print("❌ 刷新后的 Token A2 仍然有效\n")
|
|
|
|
print(f"{'='*60}")
|
|
print("验证完成")
|
|
print(f"{'='*60}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|