330 lines
11 KiB
TypeScript
330 lines
11 KiB
TypeScript
import assert from 'node:assert/strict'
|
|
import test from 'node:test'
|
|
|
|
import { WEIBO_TICKET_PREFIX, WeiboNativeConnector } from './weibo-native-connector.js'
|
|
|
|
type CapturedCall = { url: string; init: RequestInit }
|
|
|
|
function jsonResponse(payload: unknown): Response {
|
|
return new Response(JSON.stringify(payload), {
|
|
status: 200,
|
|
headers: { 'content-type': 'application/json' },
|
|
})
|
|
}
|
|
|
|
function createFetchMock(
|
|
handler: (url: string, init: RequestInit) => Response | Promise<Response>,
|
|
) {
|
|
const calls: CapturedCall[] = []
|
|
const fetchImpl = (async (input: RequestInfo | URL, init?: RequestInit) => {
|
|
const url =
|
|
typeof input === 'string'
|
|
? input
|
|
: input instanceof URL
|
|
? input.toString()
|
|
: String(input.url)
|
|
const requestInit = init || {}
|
|
calls.push({ url, init: requestInit })
|
|
return handler(url, requestInit)
|
|
}) as typeof fetch
|
|
return { fetchImpl, calls }
|
|
}
|
|
|
|
function defaultHandler(url: string): Response {
|
|
if (url.includes('/sso/signin')) {
|
|
return new Response('<html></html>', {
|
|
status: 200,
|
|
headers: {
|
|
'content-type': 'text/html',
|
|
'set-cookie': 'X-CSRF-TOKEN=CK-demo; Path=/sso',
|
|
},
|
|
})
|
|
}
|
|
if (url.includes('/sso/v2/web/config')) {
|
|
return jsonResponse({
|
|
retcode: 20000000,
|
|
data: {
|
|
country_code: [
|
|
{ country_code: '86', local: { zh_CN: '中国' } },
|
|
{ country_code: '852', local: { zh_CN: '中国香港' } },
|
|
{ country_code: '886', local: { zh_CN: '中国台湾' } },
|
|
],
|
|
},
|
|
})
|
|
}
|
|
if (url.includes('/sso/v2/qrcode/image')) {
|
|
return jsonResponse({
|
|
retcode: 20000000,
|
|
data: { qrid: 'qr-1', image: 'https://v2.qr.weibo.cn/qr-demo.png' },
|
|
})
|
|
}
|
|
if (url.includes('/sso/v2/qrcode/check')) {
|
|
return jsonResponse({ retcode: 50114001, msg: '未使用', data: null })
|
|
}
|
|
if (url.startsWith('https://v2.qr.weibo.cn/')) {
|
|
return new Response(Buffer.from('fake-png'), {
|
|
status: 200,
|
|
headers: { 'content-type': 'image/png' },
|
|
})
|
|
}
|
|
if (url.includes('/sso/v2/sms/send')) {
|
|
return jsonResponse({ retcode: 20000000 })
|
|
}
|
|
if (url.includes('/sso/v2/login')) {
|
|
const headers = new Headers({
|
|
'content-type': 'application/json',
|
|
})
|
|
headers.append('set-cookie', 'SUB=demo-sub; Path=/; Domain=.weibo.com')
|
|
headers.append('set-cookie', 'PASSPORT_ONLY=demo-passport; Path=/sso')
|
|
return new Response(
|
|
JSON.stringify({
|
|
retcode: 20000000,
|
|
data: {
|
|
data: {
|
|
act: 'goto',
|
|
location: 'https://passport.weibo.com/sso/v2/goto?ticket=demo',
|
|
},
|
|
},
|
|
}),
|
|
{
|
|
status: 200,
|
|
headers,
|
|
},
|
|
)
|
|
}
|
|
if (url.includes('/sso/v2/goto')) {
|
|
return new Response(null, {
|
|
status: 302,
|
|
headers: {
|
|
location: 'https://weibo.com/',
|
|
'set-cookie': 'SUB=demo-sub; Path=/; Domain=.weibo.com',
|
|
},
|
|
})
|
|
}
|
|
if (url === 'https://weibo.com/') {
|
|
return new Response('<html></html>', {
|
|
status: 200,
|
|
headers: { 'content-type': 'text/html' },
|
|
})
|
|
}
|
|
if (url.includes('/ajax/side/info')) {
|
|
return jsonResponse({
|
|
data: {
|
|
user: {
|
|
id: 123456,
|
|
screen_name: '微博测试账号',
|
|
avatar_large: 'https://tvax1.sinaimg.cn/demo.jpg',
|
|
},
|
|
},
|
|
})
|
|
}
|
|
return new Response('not found', { status: 404 })
|
|
}
|
|
|
|
test('微博二维码会话解析 qrid 并把图片转成 base64', async () => {
|
|
const { fetchImpl, calls } = createFetchMock(defaultHandler)
|
|
const connector = new WeiboNativeConnector({ fetchImpl })
|
|
const session = await connector.createQrLoginSession({ taskId: 1, orderNo: 'MOCKWB1' })
|
|
|
|
assert.equal(session.status, 'waiting_scan')
|
|
assert.match(session.qrImage || '', /^data:image\/png;base64,/)
|
|
assert.ok(calls.some((call) => call.url.includes('qrcode/image?entry=miniblog&size=180')))
|
|
})
|
|
|
|
test('微博二维码未扫码时保持等待扫码状态', async () => {
|
|
const { fetchImpl } = createFetchMock(defaultHandler)
|
|
const connector = new WeiboNativeConnector({ fetchImpl })
|
|
const session = await connector.createQrLoginSession({ taskId: 1, orderNo: 'MOCKWB1' })
|
|
const polled = await connector.getQrLoginSession(session.sessionId)
|
|
assert.equal(polled.status, 'waiting_scan')
|
|
})
|
|
|
|
test('微博发送短信验证码使用大陆 11 位手机号', async () => {
|
|
const { fetchImpl, calls } = createFetchMock(defaultHandler)
|
|
const connector = new WeiboNativeConnector({ fetchImpl })
|
|
await connector.sendSmsCode({ taskId: 1, mobile: '18900762182' })
|
|
|
|
const call = calls.find((item) => item.url.includes('/sso/v2/sms/send'))
|
|
assert.ok(call)
|
|
const form = new URLSearchParams(String(call.init.body))
|
|
assert.equal(form.get('entry'), 'miniblog')
|
|
assert.equal(form.get('mobile'), 'T3JH3iRZfvXRi9K')
|
|
assert.equal(form.get('mfa_id'), null)
|
|
assert.equal(form.get('el'), '2')
|
|
})
|
|
|
|
test('微博短信发送支持其他国家码并按 00 前缀编码', async () => {
|
|
const { fetchImpl, calls } = createFetchMock(defaultHandler)
|
|
const connector = new WeiboNativeConnector({ fetchImpl })
|
|
await connector.sendSmsCode({
|
|
taskId: 1,
|
|
mobile: '61234567',
|
|
countryCode: '852',
|
|
})
|
|
|
|
const call = calls.find((item) => item.url.includes('/sso/v2/sms/send'))
|
|
assert.ok(call)
|
|
const form = new URLSearchParams(String(call.init.body))
|
|
assert.equal(form.get('mobile'), 'TiNs13aS9CXDe9Nva3')
|
|
})
|
|
|
|
test('微博国家码列表来自 web config 并缓存', async () => {
|
|
const { fetchImpl, calls } = createFetchMock(defaultHandler)
|
|
const connector = new WeiboNativeConnector({ fetchImpl })
|
|
const first = await connector.getCountryCodes()
|
|
const second = await connector.getCountryCodes()
|
|
|
|
assert.deepEqual(first, [
|
|
{ code: '86', name: '中国' },
|
|
{ code: '852', name: '中国香港' },
|
|
{ code: '886', name: '中国台湾' },
|
|
])
|
|
assert.deepEqual(second, first)
|
|
assert.equal(calls.filter((call) => call.url.includes('/sso/v2/web/config')).length, 1)
|
|
})
|
|
|
|
test('微博短信登录会带上发码响应下发的 Cookie', async () => {
|
|
const { fetchImpl, calls } = createFetchMock((url) => {
|
|
if (url.includes('/sso/v2/sms/send')) {
|
|
return new Response(JSON.stringify({ retcode: 20000000 }), {
|
|
status: 200,
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'set-cookie': 'SCF=send-cookie; Path=/',
|
|
},
|
|
})
|
|
}
|
|
return defaultHandler(url)
|
|
})
|
|
const connector = new WeiboNativeConnector({ fetchImpl })
|
|
await connector.sendSmsCode({ taskId: 1, mobile: '13800138000' })
|
|
const { session } = await connector.loginWithSms({
|
|
taskId: 1,
|
|
orderNo: 'MOCKWB1',
|
|
mobile: '13800138000',
|
|
code: '123456',
|
|
})
|
|
|
|
assert.equal(session.status, 'authenticated')
|
|
const redirectCall = calls.find((item) => item.url.includes('/sso/v2/goto'))
|
|
assert.ok(redirectCall)
|
|
const headers = redirectCall.init.headers as Record<string, string>
|
|
assert.match(headers.cookie || '', /SCF=send-cookie/)
|
|
})
|
|
|
|
test('微博短信登录建立会话并可生成一次性注入票据', async () => {
|
|
const { fetchImpl, calls } = createFetchMock(defaultHandler)
|
|
const connector = new WeiboNativeConnector({ fetchImpl })
|
|
await connector.sendSmsCode({ taskId: 1, mobile: '138 0013 8000' })
|
|
const { sessionId, session } = await connector.loginWithSms({
|
|
taskId: 1,
|
|
orderNo: 'MOCKWB1',
|
|
mobile: '138 0013 8000',
|
|
code: '123456',
|
|
})
|
|
|
|
assert.equal(session.status, 'authenticated')
|
|
assert.equal(session.huyaUid, '123456')
|
|
assert.equal(session.nickname, '微博测试账号')
|
|
|
|
const loginCall = calls.find((item) => item.url.includes('/sso/v2/login'))
|
|
assert.ok(loginCall)
|
|
const form = new URLSearchParams(String(loginCall.init.body))
|
|
assert.equal(form.get('type'), '2')
|
|
assert.equal(form.get('username'), 'T3Gs3iR_9fN7A9t')
|
|
assert.equal(form.get('scode'), 'T3r81iMS')
|
|
|
|
const ticket = await connector.createBrowserTicket({
|
|
sessionId,
|
|
targetUrl: 'https://weibo.com/',
|
|
})
|
|
assert.ok(ticket.ticket.startsWith(WEIBO_TICKET_PREFIX))
|
|
|
|
const payload = await connector.consumeBrowserTicket(ticket.ticket)
|
|
assert.equal(payload.platform, 'weibo')
|
|
assert.equal(payload.targetUrl, 'https://weibo.com/')
|
|
assert.ok(payload.cookies.some((cookie) => cookie.name === 'SUB' && cookie.value === 'demo-sub'))
|
|
|
|
await assert.rejects(() => connector.consumeBrowserTicket(ticket.ticket), /票据不存在或已过期/)
|
|
})
|
|
|
|
test('微博短信发送遇到 mfa_1 时返回易盾挑战，完成挑战后可用 mfa_id 重发', async () => {
|
|
let sendCount = 0
|
|
const { fetchImpl, calls } = createFetchMock((url, init) => {
|
|
if (url.includes('/sso/v2/sms/send')) {
|
|
sendCount += 1
|
|
const form = new URLSearchParams(String(init.body))
|
|
if (!form.get('mfa_id')) {
|
|
return jsonResponse({
|
|
retcode: 0,
|
|
msg: 'need mfa',
|
|
data: { act: 'mfa_1', mfa_id: 'mfa-demo' },
|
|
})
|
|
}
|
|
return jsonResponse({ retcode: 20000000, msg: '', data: [] })
|
|
}
|
|
return defaultHandler(url)
|
|
})
|
|
const connector = new WeiboNativeConnector({ fetchImpl })
|
|
|
|
const first = await connector.sendSmsCode({ taskId: 1, mobile: '13800138000' })
|
|
assert.equal(first.mfaRequired, true)
|
|
assert.equal(first.mfaId, 'mfa-demo')
|
|
assert.equal(first.captchaId, '7cda0ba2785647ada4d6e946ddb2d465')
|
|
|
|
const second = await connector.sendSmsCode({
|
|
taskId: 1,
|
|
mobile: '13800138000',
|
|
mfaId: 'mfa-demo',
|
|
})
|
|
assert.equal(second.mfaRequired, false)
|
|
assert.equal(sendCount, 2)
|
|
const secondForm = new URLSearchParams(
|
|
String(calls.filter((call) => call.url.includes('/sso/v2/sms/send'))[1]?.init.body),
|
|
)
|
|
assert.equal(secondForm.get('mfa_id'), 'mfa-demo')
|
|
})
|
|
|
|
test('微博 CK 可按账号级凭证导出并从序列化 Cookie 恢复会话', async () => {
|
|
const { fetchImpl } = createFetchMock(defaultHandler)
|
|
const connector = new WeiboNativeConnector({ fetchImpl })
|
|
await connector.sendSmsCode({ taskId: 1, mobile: '13800138000' })
|
|
const { sessionId } = await connector.loginWithSms({
|
|
taskId: 1,
|
|
orderNo: 'MOCKWB1',
|
|
mobile: '13800138000',
|
|
code: '123456',
|
|
})
|
|
const credential = connector.exportSessionCredential(sessionId)
|
|
assert.equal(credential.platformUid, '123456')
|
|
assert.ok(credential.cookies.some((cookie) => cookie.name === 'SUB'))
|
|
assert.ok(
|
|
credential.cookies.some(
|
|
(cookie) =>
|
|
cookie.name === 'PASSPORT_ONLY' &&
|
|
cookie.domain === 'passport.weibo.com' &&
|
|
cookie.path === '/sso',
|
|
),
|
|
)
|
|
|
|
const ticket = await connector.createBrowserTicket({
|
|
sessionId,
|
|
targetUrl: 'https://weibo.com/',
|
|
})
|
|
const injection = await connector.consumeBrowserTicket(ticket.ticket)
|
|
assert.equal(
|
|
injection.cookies.some((cookie) => cookie.name === 'PASSPORT_ONLY'),
|
|
false,
|
|
)
|
|
|
|
const restored = await connector.restoreSessionFromCredential({
|
|
taskId: 2,
|
|
orderNo: 'MOCKWB2',
|
|
platformUid: credential.platformUid,
|
|
nickname: credential.nickname,
|
|
cookies: credential.cookies,
|
|
})
|
|
assert.equal(restored.status, 'authenticated')
|
|
assert.equal(restored.hasCredential, true)
|
|
})
|