From 515567f6d404b7c46cc7cad0b2ebf93f8a318b15 Mon Sep 17 00:00:00 2001 From: stu2not Date: Wed, 23 Sep 2026 17:47:42 +0800 Subject: [PATCH] feat(binding): verify 4G relations before Banban unbind --- banban-mini/src/app.config.ts | 2 + banban-mini/src/app.tsx | 3 + banban-mini/src/pages/bind/index.tsx | 5 + banban-mini/src/pages/sleep/index.tsx | 48 +- .../src/pages/traffic-binding/index.config.ts | 1 + .../src/pages/traffic-binding/index.scss | 7 + .../src/pages/traffic-binding/index.tsx | 69 +++ banban-mini/src/services/api.ts | 17 +- banban-mini/src/services/traffic-binding.ts | 157 +++++ banban-mini/tests/api-timeout.test.cjs | 31 + banban-mini/tests/traffic-binding.test.cjs | 152 +++++ ...banban-4g-cross-miniapp-handoff-binding.md | 166 ++++++ docs/设备直接绑定接口联调文档(2).md | 564 ++++++++++++++++++ talkingq-url/banban/dao/binding.py | 33 +- talkingq-url/banban/routers/__init__.py | 2 + talkingq-url/banban/routers/bindings.py | 3 +- talkingq-url/banban/routers/joint_binding.py | 74 +++ talkingq-url/banban/service/joint_binding.py | 297 +++++++++ .../banban/service/traffic_handoff.py | 89 +++ talkingq-url/config.py | 5 + talkingq-url/handlers/mqtt_handler.py | 18 +- talkingq-url/mysql/init/02-init.sql | 4 + talkingq-url/mysql/init/03-joint-binding.sql | 17 + .../migrations/20260923_joint_binding.sql | 24 + talkingq-url/tests/test_joint_binding.py | 347 +++++++++++ talkingq-url/tests/test_traffic_relation.py | 79 +++ 26 files changed, 2185 insertions(+), 29 deletions(-) create mode 100644 banban-mini/src/pages/traffic-binding/index.config.ts create mode 100644 banban-mini/src/pages/traffic-binding/index.scss create mode 100644 banban-mini/src/pages/traffic-binding/index.tsx create mode 100644 banban-mini/src/services/traffic-binding.ts create mode 100644 banban-mini/tests/api-timeout.test.cjs create mode 100644 banban-mini/tests/traffic-binding.test.cjs create mode 100644 docs/banban-4g-cross-miniapp-handoff-binding.md create mode 100644 docs/设备直接绑定接口联调文档(2).md create mode 100644 talkingq-url/banban/routers/joint_binding.py create mode 100644 talkingq-url/banban/service/joint_binding.py create mode 100644 talkingq-url/banban/service/traffic_handoff.py create mode 100644 talkingq-url/mysql/init/03-joint-binding.sql create mode 100644 talkingq-url/mysql/migrations/20260923_joint_binding.sql create mode 100644 talkingq-url/tests/test_joint_binding.py create mode 100644 talkingq-url/tests/test_traffic_relation.py diff --git a/banban-mini/src/app.config.ts b/banban-mini/src/app.config.ts index 7eb45b7..0679cf0 100644 --- a/banban-mini/src/app.config.ts +++ b/banban-mini/src/app.config.ts @@ -3,6 +3,7 @@ export default defineAppConfig({ "pages/device/index", "pages/login/index", "pages/bind/index", + "pages/traffic-binding/index", "pages/notification/index", "pages/chat/index", "pages/chat/detail/index", @@ -12,6 +13,7 @@ export default defineAppConfig({ "pages/family-invite/index", "pages/webview/index", ], + navigateToMiniProgramAppIdList: ['wx56e1cfc52aab630f'], window: { backgroundTextStyle: "light", navigationBarBackgroundColor: "#FFF9EA", diff --git a/banban-mini/src/app.tsx b/banban-mini/src/app.tsx index 143d6b2..78e19b4 100644 --- a/banban-mini/src/app.tsx +++ b/banban-mini/src/app.tsx @@ -1,7 +1,10 @@ import './app.scss' import { PropsWithChildren } from 'react' +import { useDidShow } from '@tarojs/taro' +import { captureTrafficReturn } from './services/traffic-binding' function App(props: PropsWithChildren) { + useDidShow((options) => { captureTrafficReturn(options) }) return props.children } diff --git a/banban-mini/src/pages/bind/index.tsx b/banban-mini/src/pages/bind/index.tsx index a0119b9..6bb597c 100644 --- a/banban-mini/src/pages/bind/index.tsx +++ b/banban-mini/src/pages/bind/index.tsx @@ -26,6 +26,7 @@ import { validateChildNameInput, } from '@/services/child' import { useSystemBanner } from '@/components/system-banner/use-system-banner' +import { trafficPageUrl } from '@/services/traffic-binding' import './index.scss' @@ -678,6 +679,10 @@ export default function Bind() { + {!isAddingCard && } + {!isAddingCard && 设备已绑定。流量查询和充值需另外关联，可稍后在管理页完成。} ) : ( <> diff --git a/banban-mini/src/pages/sleep/index.tsx b/banban-mini/src/pages/sleep/index.tsx index da01770..f15f80e 100644 --- a/banban-mini/src/pages/sleep/index.tsx +++ b/banban-mini/src/pages/sleep/index.tsx @@ -10,6 +10,7 @@ import Taro, { useDidHide, useDidShow, useShareAppMessage } from '@tarojs/taro' import { authorizeParentPhone, clearToken, getParent, getToken, Parent } from '@/services/auth' import { DEVICE_UNAVAILABLE_MESSAGE } from '@/services/api' import { getCurrentUserId } from '@/services/session' +import { TrafficOverview, getTrafficOverview, trafficLinked, openTrafficPage, openTrafficPortal, flushTrafficReceipt } from '@/services/traffic-binding' import { BindingListItem, BindingCardItem, @@ -19,7 +20,6 @@ import { loadCurrentChildBindingContext, setSelectedBindingDeviceId, updateBindingCardName, - unbindDevice, } from '@/services/binding' import { Child, @@ -77,6 +77,20 @@ export default function Sleep() { const [children, setChildren] = useState([]) const [bindings, setBindings] = useState([]) const [binding, setBinding] = useState(null) + const [trafficOverview, setTrafficOverview] = useState(null) + const [trafficBusy, setTrafficBusy] = useState(false) + const trafficLock = useRef(false) + useEffect(() => { + let cancelled = false + setTrafficOverview(null) + if (binding?.device_id) { + void flushTrafficReceipt().catch(() => null) + .then(() => getTrafficOverview(binding.device_id)) + .then((value) => { if (!cancelled) setTrafficOverview(value) }) + .catch(() => { /* Core device management remains available. */ }) + } + return () => { cancelled = true } + }, [binding]) const [currentChild, setCurrentChild] = useState(null) const [firmwareStatus, setFirmwareStatus] = useState(null) const [showFirmwareProgress, setShowFirmwareProgress] = useState(false) @@ -651,12 +665,7 @@ export default function Sleep() { if (!result.confirm) return try { - await unbindDevice(targetBinding.device_id) - if (binding?.device_id === targetBinding.device_id) { - clearSelectedBindingDeviceId() - } - Taro.showToast({ title: '已删除设备', icon: 'success' }) - await loadData() + await openTrafficPage(targetBinding.device_id, 'unbind') } catch (error: any) { Taro.showToast({ title: error?.message || '删除设备失败，请重试', @@ -834,10 +843,7 @@ export default function Sleep() { if (!result.confirm) return try { - await unbindDevice(binding.device_id) - clearSelectedBindingDeviceId() - Taro.showToast({ title: '已解绑', icon: 'success' }) - await loadData() + await openTrafficPage(binding.device_id, 'unbind') } catch (error: any) { Taro.showToast({ title: error?.message || '解绑失败，请重试', @@ -956,6 +962,18 @@ export default function Sleep() { return } + if (['关联流量服务', '继续确认流量服务', '继续解绑', '流量查询', '套餐充值'].includes(item.name)) { + if (!binding || trafficLock.current) return + trafficLock.current = true + setTrafficBusy(true) + const operation = item.name === '流量查询' || item.name === '套餐充值' + ? openTrafficPortal(binding.device_id, item.name === '套餐充值' ? 'recharge' : 'device') + : openTrafficPage(binding.device_id, item.name === '继续解绑' ? 'unbind' : 'bind') + void operation.catch((error: any) => Taro.showToast({ title: error.message || error.errMsg || '操作未完成，请重试', icon: 'none' })) + .finally(() => { trafficLock.current = false; setTrafficBusy(false) }) + return + } + if (item.name === '我的卡片') { setShowCardsModal(true) return @@ -1129,6 +1147,14 @@ export default function Sleep() { value: binding?.device_id ? `当前: ${binding.device_id}` : '未绑定', arrow: true, }, + ...(binding ? (trafficLinked(trafficOverview) && trafficOverview?.device_id === binding.device_id + ? [ + { icon: require('../../assets/ui/menu-binding.png'), iconBgClass: 'green', name: '流量查询', value: '', arrow: true, disabled: !isFamilyOwner || trafficBusy }, + { icon: require('../../assets/ui/menu-binding.png'), iconBgClass: 'green', name: '套餐充值', value: '', arrow: true, disabled: !isFamilyOwner || trafficBusy }, + ] + : [{ icon: require('../../assets/ui/menu-binding.png'), iconBgClass: 'green', + name: trafficOverview?.unbind ? '继续解绑' : trafficOverview?.bind ? '继续确认流量服务' : '关联流量服务', + value: isFamilyOwner ? '不影响伴伴使用' : '仅设备归属账号可操作', arrow: true, disabled: !isFamilyOwner || trafficBusy }]) : []), { icon: require('../../assets/ui/menu-cards.png'), iconBgClass: 'green', diff --git a/banban-mini/src/pages/traffic-binding/index.config.ts b/banban-mini/src/pages/traffic-binding/index.config.ts new file mode 100644 index 0000000..b555be4 --- /dev/null +++ b/banban-mini/src/pages/traffic-binding/index.config.ts @@ -0,0 +1 @@ +export default definePageConfig({ navigationBarTitleText: '流量服务关联' }) diff --git a/banban-mini/src/pages/traffic-binding/index.scss b/banban-mini/src/pages/traffic-binding/index.scss new file mode 100644 index 0000000..9ffee51 --- /dev/null +++ b/banban-mini/src/pages/traffic-binding/index.scss @@ -0,0 +1,7 @@ +.traffic-binding-page { + min-height: 100vh; box-sizing: border-box; padding: 72rpx 40rpx; + .title { display: block; font-size: 38rpx; font-weight: bold; margin-bottom: 30rpx; } + .description { display: block; font-size: 28rpx; line-height: 1.7; margin-bottom: 30rpx; } + .error { display: block; color: #a52a2a; margin-bottom: 24rpx; } + button { margin-top: 24rpx; } +} diff --git a/banban-mini/src/pages/traffic-binding/index.tsx b/banban-mini/src/pages/traffic-binding/index.tsx new file mode 100644 index 0000000..f07bea5 --- /dev/null +++ b/banban-mini/src/pages/traffic-binding/index.tsx @@ -0,0 +1,69 @@ +import { View, Text, Button } from '@tarojs/components' +import Taro, { useDidShow, useRouter } from '@tarojs/taro' +import { useRef, useState } from 'react' +import { getToken } from '@/services/auth' +import { clearSelectedBindingDeviceId } from '@/services/binding' +import { getPageStyle } from '@/utils/page-style' +import { TrafficOperation, getTrafficOperation, resumeTrafficOperation, + flushTrafficReceipt, finishTrafficUnbind, jumpTraffic } from '@/services/traffic-binding' +import './index.scss' + +export default function TrafficBindingPage() { + const { params } = useRouter() + const deviceId = String(params.device_id || '') + const action = params.action === 'unbind' ? 'unbind' : 'bind' + const [operation, setOperation] = useState(null) + const [busy, setBusy] = useState(false) + const [error, setError] = useState('') + const lock = useRef(false) + + const recover = async () => { + if (lock.current) return + if (!getToken()) { Taro.navigateTo({ url: '/pages/login/index' }); return } + lock.current = true + setBusy(true) + try { + const current = await resumeTrafficOperation(deviceId, action) + setOperation(current) + if (action === 'unbind' && current.status === 'completed') clearSelectedBindingDeviceId() + setError('') + } catch (e: any) { setError(e.message || '暂时无法确认，请重试') } + finally { lock.current = false; setBusy(false) } + } + useDidShow(() => { void recover() }) + + const proceed = async () => { + if (!operation) { await recover(); return } + if (lock.current) return + lock.current = true + setBusy(true) + setError('') + try { + await flushTrafficReceipt() + let current = await getTrafficOperation(operation.operation_id) + if (current.status !== 'completed') { + const result = await jumpTraffic(current) + if (!result) return + current = result + if (action === 'unbind') current = await finishTrafficUnbind(current) + } + setOperation(current) + if (action === 'unbind' && current.status === 'completed') clearSelectedBindingDeviceId() + } catch (e: any) { setError(e.message || e.errMsg || '操作未完成，请继续重试') } + finally { lock.current = false; setBusy(false) } + } + const completed = operation?.status === 'completed' + const finalUnbind = action === 'unbind' && operation?.traffic_status === 'server_confirmed' + return + {action === 'bind' ? completed ? '流量服务已关联' : '设备已绑定，流量服务待关联' : completed ? '设备已解除绑定' : '解除设备与流量服务绑定'} + {action === 'bind' + ? completed ? '可以查询流量和购买套餐了。' : '伴伴设备可以正常使用。继续前往流量管理小程序登录并关联，也可以稍后完成。' + : completed ? '设备已解绑，流量侧已确认没有本次有效绑定。' : finalUnbind ? '流量后端已确认可以解绑，点击完成伴伴解绑。' : '先确认流量侧原绑定已解除，再解除伴伴绑定；套餐和充值记录保留。继续处理会停止本次流量关联，暂不支持取消恢复。'} + {operation?.last_error === 'HANDOFF_RELATION_UNKNOWN' && 流量后端无法确认原绑定记录，伴伴绑定仍保留，请联系服务人员核查。} + {operation?.last_error === 'TRAFFIC_UNBIND_REQUIRED' && 原流量绑定仍有效，请继续前往流量管理小程序解绑。} + {operation?.status === 'partial_failed' && 上次流量操作未完成，可以继续重试。} + {error && {error}} + {!completed && } + + +} diff --git a/banban-mini/src/services/api.ts b/banban-mini/src/services/api.ts index 1e2ad10..7891a54 100644 --- a/banban-mini/src/services/api.ts +++ b/banban-mini/src/services/api.ts @@ -12,7 +12,7 @@ const ERROR_DETAIL_MAP: Record = { } class ApiError extends Error { - constructor(public status: number, message: string) { + constructor(public status: number, message: string, public code?: string) { super(message) } } @@ -21,6 +21,7 @@ function getErrorDetail(data: any): string { if (!data) return ERROR_DETAIL_MAP['Request failed'] if (typeof data.errMsg === 'string') return data.errMsg if (typeof data.detail === 'string') return ERROR_DETAIL_MAP[data.detail] || data.detail + if (typeof data.detail?.message === 'string') return data.detail.message if (Array.isArray(data.detail)) { return data.detail.map((item: any) => String(item?.msg || item)).join('; ') } @@ -48,25 +49,33 @@ async function request( } let response + let timer: ReturnType | undefined try { - response = await Taro.request({ + const task = Taro.request({ url: requestUrl, method: requestMethod, data: options.data, header: headers, timeout: REQUEST_TIMEOUT_MS, }) + response = await Promise.race([task, new Promise((_, reject) => { + timer = setTimeout(() => { + reject(new Error('请求超时，请重试')) + task.abort?.() + }, REQUEST_TIMEOUT_MS) + })]) } catch (error: any) { const message = String(error?.errMsg || error?.message || 'request:fail') throw new ApiError(0, message) - } + } finally { if (timer) clearTimeout(timer) } if (response.statusCode === 401) { handleUnauthorized({ redirect: !url.startsWith('/banban/auth/') }) } if (response.statusCode >= 400) { - throw new ApiError(response.statusCode, getErrorDetail(response.data)) + const body = response.data as any + throw new ApiError(response.statusCode, getErrorDetail(body), body?.detail?.code || body?.code) } return response.data as T diff --git a/banban-mini/src/services/traffic-binding.ts b/banban-mini/src/services/traffic-binding.ts new file mode 100644 index 0000000..b7fa622 --- /dev/null +++ b/banban-mini/src/services/traffic-binding.ts @@ -0,0 +1,157 @@ +import Taro from '@tarojs/taro' +import { request } from './api' +import { getCurrentUserId } from './session' + +export const TRAFFIC_APP_ID = 'wx56e1cfc52aab630f' +export type TrafficAction = 'bind' | 'unbind' +export interface TrafficOperation { + operation_id: string + device_id: string + action: TrafficAction + status: 'pending' | 'partial_failed' | 'completed' + traffic_status: 'pending' | 'client_confirmed' | 'server_confirmed' + last_error?: string | null +} +export interface TrafficOverview { + device_id: string + binding_instance_id: string + bind: TrafficOperation | null + unbind: TrafficOperation | null +} +const key = () => `trafficBinding:${getCurrentUserId()}` +type Pending = { operation: TrafficOperation; receipt?: any } + +export function getPendingTraffic(): Pending | null { + if (!getCurrentUserId()) return null + const value = Taro.getStorageSync(key()) + return value?.operation?.operation_id ? value : null +} +export function rememberOperation(operation: TrafficOperation) { + if (!getCurrentUserId()) return + const previous = getPendingTraffic() + Taro.setStorageSync(key(), { operation, receipt: previous?.operation.operation_id === operation.operation_id ? previous.receipt : undefined }) +} +export function captureTrafficReturn(options: any): boolean { + if (options?.referrerInfo?.appId !== TRAFFIC_APP_ID) return false + const data = options.referrerInfo.extraData + const pending = getPendingTraffic() + if (!pending || !data || data.operationId !== pending.operation.operation_id + || data.action !== pending.operation.action || data.requestId !== `joint-${data.operationId}` + || typeof data.success !== 'boolean' || !/^[A-Za-z0-9_]{1,64}$/.test(data.code || '')) return false + if (data.success && (data.code !== 'OK' || !/^[A-Z0-9]{18,22}$/.test(data.iccid || ''))) return false + // Only store the bounded receipt, never another app's tokens or arbitrary extraData. + Taro.setStorageSync(key(), { operation: pending.operation, receipt: { + action: data.action, request_id: data.requestId, success: data.success, code: data.code, + iccid: data.success ? data.iccid : null, + } }) + return true +} +export function trafficLinked(overview: TrafficOverview | null): boolean { + return Boolean(overview?.bind?.status === 'completed' && overview.bind.traffic_status === 'server_confirmed' && !overview?.unbind) +} +export async function getTrafficOverview(deviceId: string) { + const user = getCurrentUserId() + const overview = await request(`/banban/joint-bindings/devices/${encodeURIComponent(deviceId)}`) + assertSameUser(user) + if (overview.bind && !overview.unbind) overview.bind = await checkTrafficOperation(overview.bind.operation_id) + return overview +} +export function getTrafficOperation(operationId: string) { + return request(`/banban/joint-bindings/${encodeURIComponent(operationId)}`) +} +export async function checkTrafficOperation(operationId: string) { + const user = getCurrentUserId() + const operation = await request(`/banban/joint-bindings/${encodeURIComponent(operationId)}/check`, { method: 'POST' }) + assertSameUser(user) + return operation +} +export async function startTrafficOperation(deviceId: string, action: TrafficAction) { + const user = getCurrentUserId() + const operation = await request('/banban/joint-bindings', { method: 'POST', data: { device_id: deviceId, action } }) + assertSameUser(user) + rememberOperation(operation) + return operation +} +export async function flushTrafficReceipt(): Promise { + const pending = getPendingTraffic() + if (!pending?.receipt) return null + const user = getCurrentUserId() + let result: TrafficOperation + try { + result = await request(`/banban/joint-bindings/${pending.operation.operation_id}/result`, { + method: 'POST', data: pending.receipt, + }) + } catch (error: any) { + assertSameUser(user) + if (!['STALE_OPERATION', 'OPERATION_NOT_FOUND'].includes(error.code)) throw error + // A previous binding instance must not prevent this device from being bound again. + if (getPendingTraffic()?.operation.operation_id === pending.operation.operation_id) Taro.setStorageSync(key(), null) + return null + } + assertSameUser(user) + if (getPendingTraffic()?.operation.operation_id === result.operation_id) Taro.setStorageSync(key(), { operation: result }) + return result +} +export async function resumeTrafficOperation(deviceId: string, action: TrafficAction) { + await flushTrafficReceipt() + const pending = getPendingTraffic() + if (pending?.operation.device_id === deviceId && pending.operation.action === action) { + try { return await checkTrafficOperation(pending.operation.operation_id) } + catch (error: any) { + if (!['STALE_OPERATION', 'OPERATION_NOT_FOUND'].includes(error.code)) throw error + } + } + return startTrafficOperation(deviceId, action) +} +export async function finishTrafficUnbind(operation: TrafficOperation) { + const user = getCurrentUserId() + const result = await request(`/banban/joint-bindings/${operation.operation_id}/finish-unbind`, { method: 'POST' }) + assertSameUser(user) + rememberOperation(result) + return result +} +function targetVersion(): 'trial' | 'release' { + return Taro.getAccountInfoSync().miniProgram.envVersion === 'release' ? 'release' : 'trial' +} +function assertSameUser(user: number) { + if (!user || getCurrentUserId() !== user) throw new Error('账号已变化，请重新进入') +} +async function navigate(options: any, withinApp = false) { + let timer: ReturnType | undefined + try { + return await Promise.race([withinApp ? Taro.navigateTo(options) : Taro.navigateToMiniProgram(options), new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error('跳转等待已结束，请重新进入确认')), 15000) + })]) + } finally { if (timer) clearTimeout(timer) } +} +export async function jumpTraffic(operation: TrafficOperation) { + const user = getCurrentUserId() + const data = await request(`/banban/joint-bindings/${operation.operation_id}/prepare`, { method: 'POST' }) + assertSameUser(user) + rememberOperation(data.operation) + if (data.operation.status === 'completed' || data.operation.traffic_status === 'server_confirmed') return data.operation as TrafficOperation + if (data.target_app_id !== TRAFFIC_APP_ID) throw new Error('目标小程序配置不一致') + await navigate({ + appId: TRAFFIC_APP_ID, path: 'pages/handoff/index', envVersion: targetVersion(), + extraData: { operationId: operation.operation_id, action: operation.action, + requestId: data.request_id, sourceBindingId: data.source_binding_id, + handoffToken: data.handoff_token, iccid: data.iccid, deviceId: operation.device_id, + recoveryOnly: data.recovery_only === true }, + }) + return null +} +export async function openTrafficPortal(deviceId: string, destination: 'device' | 'recharge') { + const user = getCurrentUserId() + const data = await request(`/banban/joint-bindings/devices/${encodeURIComponent(deviceId)}/portal`) + assertSameUser(user) + if (data.target_app_id !== TRAFFIC_APP_ID) throw new Error('目标小程序配置不一致') + await navigate({ appId: TRAFFIC_APP_ID, path: 'pages/handoff/index', envVersion: targetVersion(), + extraData: { action: 'view', destination, iccid: data.iccid, deviceId, + sourceBindingId: data.source_binding_id } }) +} +export function trafficPageUrl(deviceId: string, action: TrafficAction = 'bind') { + return `/pages/traffic-binding/index?device_id=${encodeURIComponent(deviceId)}&action=${action}` +} +export function openTrafficPage(deviceId: string, action: TrafficAction = 'bind') { + return navigate({ url: trafficPageUrl(deviceId, action) }, true) +} diff --git a/banban-mini/tests/api-timeout.test.cjs b/banban-mini/tests/api-timeout.test.cjs new file mode 100644 index 0000000..3ad94d0 --- /dev/null +++ b/banban-mini/tests/api-timeout.test.cjs @@ -0,0 +1,31 @@ +const assert = require('node:assert/strict') +const { test } = require('node:test') +const fs = require('node:fs') +const path = require('node:path') +const ts = require('typescript') + +test('request with no platform callback ends at 10 seconds and aborts without retrying', async () => { + const timers = new Map() + let calls = 0 + let aborted = 0 + let resolve + const task = new Promise(r => { resolve = r }) + task.abort = () => { aborted++ } + const module = { exports: {} } + const source = fs.readFileSync(path.join(__dirname, '../src/services/api.ts'), 'utf8') + const js = ts.transpileModule(source, { compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2019, esModuleInterop: true } }).outputText + new Function('require', 'module', 'exports', 'setTimeout', 'clearTimeout', js)(name => { + if (name === '@tarojs/taro') return { request: () => { calls++; return task } } + if (name === '@/config/env') return { API_BASE_URL: 'https://example.invalid' } + return { getToken: () => 'test', getCurrentUserId: () => 7, handleUnauthorized: () => {} } + }, module, module.exports, (fn, ms) => { timers.set(fn, ms); return fn }, fn => timers.delete(fn)) + const pending = module.exports.request('/test') + const [timer, duration] = [...timers][0] + assert.equal(duration, 10000) + timer() + await assert.rejects(pending, /请求超时/) + assert.equal(aborted, 1) + assert.equal(calls, 1) + assert.equal(timers.size, 0) + resolve({ statusCode: 200, data: {} }) +}) diff --git a/banban-mini/tests/traffic-binding.test.cjs b/banban-mini/tests/traffic-binding.test.cjs new file mode 100644 index 0000000..7e52dd2 --- /dev/null +++ b/banban-mini/tests/traffic-binding.test.cjs @@ -0,0 +1,152 @@ +const assert = require('node:assert/strict') +const fs = require('node:fs') +const path = require('node:path') +const { test } = require('node:test') +const ts = require('typescript') +const operation = { operation_id: '12345678-1234-1234-1234-123456789abc', device_id: 'test-device', action: 'bind', status: 'pending', traffic_status: 'pending' } +const iccid = '898604E91123D1334901' + +function load() { + const storage = new Map() + let user = 7 + let version = 'trial' + const calls = [] + const jumps = [] + const timers = new Map() + let request = async (url, options) => { calls.push({ url, options }); return { ...operation, status: 'completed' } } + const Taro = { getStorageSync: k => storage.get(k), setStorageSync: (k, v) => storage.set(k, v), + navigateTo: () => new Promise(() => {}), + getAccountInfoSync: () => ({ miniProgram: { envVersion: version } }), + navigateToMiniProgram: async options => { jumps.push(options) } } + const source = fs.readFileSync(path.join(__dirname, '../src/services/traffic-binding.ts'), 'utf8') + const js = ts.transpileModule(source, { compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2019, esModuleInterop: true } }).outputText + const module = { exports: {} } + new Function('require', 'module', 'exports', 'setTimeout', 'clearTimeout', js)(id => { + if (id === '@tarojs/taro') return Taro + if (id === './session') return { getCurrentUserId: () => user } + if (id === './api') return { request: (...args) => request(...args) } + throw Error(id) + }, module, module.exports, (fn, ms) => { timers.set(fn, ms); return fn }, fn => timers.delete(fn)) + return { api: module.exports, storage, calls, jumps, timers, + user: value => { user = value }, version: value => { version = value }, request: fn => { request = fn } } +} +function returned(data = {}, appId = 'wx56e1cfc52aab630f') { + return { referrerInfo: { appId, extraData: { operationId: operation.operation_id, action: 'bind', + requestId: `joint-${operation.operation_id}`, code: 'OK', success: true, iccid, ...data } } } +} +test('only matching source, operation, action and business request can confirm', () => { + const { api } = load() + api.rememberOperation(operation) + assert.equal(api.captureTrafficReturn(returned({}, 'wrong')), false) + for (const invalid of [{ operationId: 'old' }, { action: 'unbind' }, { requestId: 'wrong' }, { success: 'true' }, { iccid: 'short' }]) { + assert.equal(api.captureTrafficReturn(returned(invalid)), false) + } + assert.equal(api.captureTrafficReturn(returned()), true) +}) +test('receipt survives network failure and account switch cannot consume it', async () => { + const context = load() + context.api.rememberOperation(operation) + context.api.captureTrafficReturn(returned()) + context.request(async () => { throw new Error('network') }) + await assert.rejects(context.api.flushTrafficReceipt(), /network/) + assert.ok(context.api.getPendingTraffic().receipt) + context.user(8) + assert.equal(context.api.getPendingTraffic(), null) + assert.equal(context.api.captureTrafficReturn(returned()), false) + context.user(7) + context.request(async () => ({ ...operation, status: 'completed' })) + await context.api.flushTrafficReceipt() + assert.equal(context.api.getPendingTraffic().receipt, undefined) +}) +test('only confirmed association opens query and recharge, unbinding closes them', () => { + const { api } = load() + assert.equal(api.trafficLinked(null), false) + assert.equal(api.trafficLinked({ bind: operation }), false) + assert.equal(api.trafficLinked({ bind: { ...operation, status: 'completed', traffic_status: 'client_confirmed' }, unbind: null }), false) + assert.equal(api.trafficLinked({ bind: { ...operation, status: 'completed', traffic_status: 'server_confirmed' }, unbind: null }), true) + assert.equal(api.trafficLinked({ bind: { ...operation, status: 'completed' }, unbind: { status: 'pending' } }), false) +}) +test('jump uses trial for testing and release in production and never treats navigation as binding', async () => { + const context = load() + context.request(async () => ({ operation, target_app_id: 'wx56e1cfc52aab630f', handoff_token: 'secret', request_id: `joint-${operation.operation_id}`, source_binding_id: 'banban:test' })) + assert.equal(await context.api.jumpTraffic(operation), null) + assert.equal(context.jumps[0].envVersion, 'trial') + assert.equal(context.api.getPendingTraffic().operation.status, 'pending') + assert.ok(!JSON.stringify([...context.storage.values()]).includes('secret')) + context.version('release') + await context.api.jumpTraffic(operation) + assert.equal(context.jumps[1].envVersion, 'release') + assert.equal(context.timers.size, 0) +}) +test('a late response after account change cannot start a jump or overwrite state', async () => { + const context = load() + context.request(async () => { context.user(8); return { operation } }) + await assert.rejects(context.api.jumpTraffic(operation), /账号已变化/) + assert.equal(context.jumps.length, 0) + assert.equal(context.api.getPendingTraffic(), null) +}) +test('rebind discards a stale receipt and starts a current operation once', async () => { + const context = load() + context.api.rememberOperation(operation) + context.api.captureTrafficReturn(returned()) + const urls = [] + context.request(async url => { + urls.push(url) + if (url.endsWith('/result')) throw Object.assign(new Error('stale'), { code: 'STALE_OPERATION' }) + return { ...operation, operation_id: 'new-instance' } + }) + const resumed = await context.api.resumeTrafficOperation(operation.device_id, 'bind') + assert.equal(resumed.operation_id, 'new-instance') + assert.equal(urls.length, 2) + assert.equal(context.api.getPendingTraffic().receipt, undefined) +}) +test('stale cached operation recovers once; network failure keeps its receipt for retry', async () => { + const context = load() + context.api.rememberOperation(operation) + context.request(async url => { + if (url.endsWith(`${operation.operation_id}/check`)) throw Object.assign(new Error('stale'), { code: 'STALE_OPERATION' }) + return { ...operation, operation_id: 'new-instance' } + }) + assert.equal((await context.api.resumeTrafficOperation(operation.device_id, 'bind')).operation_id, 'new-instance') + context.api.rememberOperation(operation) + context.api.captureTrafficReturn(returned()) + context.request(async () => { throw new Error('network') }) + await assert.rejects(context.api.resumeTrafficOperation(operation.device_id, 'bind'), /network/) + assert.ok(context.api.getPendingTraffic().receipt) +}) +test('missing navigation callback cannot keep the management button locked forever', async () => { + const context = load() + const pending = context.api.openTrafficPage(operation.device_id, 'unbind') + const [timer, duration] = [...context.timers][0] + assert.equal(duration, 15000) + timer() + await assert.rejects(pending, /跳转等待已结束/) + assert.equal(context.timers.size, 0) +}) +test('server-confirmed release proceeds locally without a redundant 4G jump', async () => { + const context = load() + const released = { ...operation, action: 'unbind', traffic_status: 'server_confirmed' } + context.request(async () => ({ operation: released })) + assert.deepEqual(await context.api.jumpTraffic(released), released) + assert.equal(context.jumps.length, 0) +}) +test('a legacy client receipt alone cannot skip the 4G result check', async () => { + const context = load() + const pending = { ...operation, action: 'unbind', traffic_status: 'client_confirmed' } + context.request(async () => ({ operation: pending, target_app_id: 'wx56e1cfc52aab630f', iccid, + request_id: `joint-${operation.operation_id}`, source_binding_id: 'banban:test' })) + assert.equal(await context.api.jumpTraffic(pending), null) + assert.equal(context.jumps.length, 1) +}) +test('management refresh checks the original relation before showing traffic entrances', async () => { + const context = load() + const calls = [] + context.request(async url => { + calls.push(url) + return url.endsWith('/check') ? { ...operation, status: 'pending', traffic_status: 'pending' } + : { device_id: operation.device_id, bind: { ...operation, status: 'completed', traffic_status: 'server_confirmed' }, unbind: null } + }) + const overview = await context.api.getTrafficOverview(operation.device_id) + assert.equal(calls.length, 2) + assert.equal(context.api.trafficLinked(overview), false) +}) diff --git a/docs/banban-4g-cross-miniapp-handoff-binding.md b/docs/banban-4g-cross-miniapp-handoff-binding.md new file mode 100644 index 0000000..5cb51cb --- /dev/null +++ b/docs/banban-4g-cross-miniapp-handoff-binding.md @@ -0,0 +1,166 @@ +# Banban 与 4G 小程序流量关联、联合解绑联调说明 + +当前契约以 [4G 后端接口文档 (2)](./设备直接绑定接口联调文档(2).md) 为准。本文描述 Banban 后端、Banban 小程序及 `/Users/stu2not/4g-miniprogram` 的本地实现；第 10 节为此前版本的线上操作记录，不表示新版关系查询/关闭流程已部署。 + +## 1. 业务规则 + +- Banban NFC 绑定成功即可正常使用设备；4G 关联允许失败、取消或稍后处理。 +- 管理页查询原 4G 关系，只有后端确认关联有效才显示查询、充值入口。 +- 用户确认联合解绑后，持久化解绑操作，阻止此绑定实例继续申请 handoff。 +- Banban 必须调用 4G 关系关闭接口，并取得 `canUnbind=true` 才能解除本地设备绑定。 +- 4G 已在其他入口解绑时，可直接确认后完成 Banban 解绑，不必再次跳转小程序。 +- `unknown`、403、404、网络失败和不存在记录均不能作为已解绑依据。 +- 不把客户端回报或当前用户设备列表中存在同一 ICCID 当作原关系有效的证明。 + +## 2. 身份与绑定实例 + +| 小程序 | AppID | +| --- | --- | +| Banban | `wx29ce99ce96b56cd9` | +| 4G 设备管理 | `wx56e1cfc52aab630f` | + +两个小程序分别登录。服务密钥仅供后端使用，不进入小程序、页面参数或日志。开发版/体验版跳转目标体验版，正式版跳转目标正式版，两侧需协调发布。 + +设备每次重新建立 Banban 绑定时生成新的 `device_bindings.binding_instance_id`。加卡、关联宝贝保持原实例。原有 `sourceBindingId=banban:{binding_instance_id}` 保持稳定，查询/关闭路径对整个 ID 做 URL 编码，包括冒号；不能为旧关系换 ID 绕过关闭状态。 + +## 3. ICCID 来源与保存 + +设备成功的 `006` 回复： + +```json +{ + "msg_id": "006", + "status": "success", + "data": { + "uuid": "53D22B6DA20001", + "iccid": "898604E91123D1334901" + } +} +``` + +ICCID 为 18～22 位字母或数字，归一化为大写，持久化到 `device_settings.iccid`。准备操作时保存到操作表作为快照。不能从 IMEI 推导；设备换卡或其他设备占用同一 ICCID 时不覆盖原记录。 + +保存最多等待 2 秒；缺失、非法或保存失败不影响 NFC 绑定。日志记录 `006 ICCID present=..., valid=...`，保存成功记录 `006 ICCID persisted`，不打印完整 ICCID。当前仅接入 `006`，没有替固件实现上报或为历史设备自动回填。 + +## 4. Banban 接口与状态 + +所有接口使用 Banban 用户登录态，并校验设备归属账号。家庭成员不能执行关联、确认或解绑。 + +| 方法与路径 | 用途 | +| --- | --- | +| `GET /banban/joint-bindings/devices/{device_id}` | 本地当前绑定实例的操作概况 | +| `POST /banban/joint-bindings` | 创建/复用 `{device_id,action:bind/unbind}` 操作 | +| `GET /banban/joint-bindings/{operation_id}` | 读取本地操作 | +| `POST /banban/joint-bindings/{operation_id}/check` | 向 4G 查询原关系并更新本地确认状态 | +| `POST /banban/joint-bindings/{operation_id}/prepare` | 绑定申请 handoff；解绑先关闭关系，必要时返回 4G 跳转信息 | +| `POST /banban/joint-bindings/{operation_id}/result` | 接收客户端结果；成功仍需服务端查询确认 | +| `POST /banban/joint-bindings/{operation_id}/finish-unbind` | 重新确认关闭结果后，事务性完成 Banban 解绑 | +| `GET /banban/joint-bindings/devices/{device_id}/portal` | 查询原关系后返回查询/充值入口 | + +原 `DELETE /banban/bindings/{device_id}` 返回 409，引导联合解绑。旧客户端需要更新才能使用新解绑流程。 + +操作的 `status=pending/partial_failed/completed` 不代表 Banban 是否已绑定。新增 `traffic_status=server_confirmed` 表示 4G 服务端关系查询/关闭已确认；旧的 `client_confirmed` 只代表客户端回报，不能授权解绑。字段已为 VARCHAR，无需为本次新增状态再加列。 + +绑定确认要求原关系为 `bound + open`。解绑确认要求 4G 明确返回 `canUnbind=true`，同时校验 `closing`、无待处理凭证、不再接受 handoff、状态为 `unbound/never_bound`。响应必须 HTTP 200、业务码 OK、sourceBindingId 匹配，布尔字段必须是布尔值，checkedAt 必须是合法 UTC 时间。 + +## 5. 流量关联流程 + +1. Banban NFC 绑定独立完成，用户自愿继续关联。 +2. 后端使用服务凭证向 `POST /internal/v1/handoff-tokens` 申请凭证；sourceBindingId 始终对应同一 ICCID。 +3. 4G 小程序使用自身登录态调用 `POST /api/v1/program/bindings/handoff`；原消费请求重试必须沿用 token 与业务 requestId。 +4. 客户端返回经过字段匹配的结果，Banban 再使用 GET 关系接口确认原绑定，确认后显示查询/充值入口。 +5. 申请返回 `HANDOFF_TOKEN_USED` 时，Banban 直接查询原关系；只有原绑定仍有效才恢复成功，不再跳转后按当前 ICCID 设备列表猜测。 +6. 4G 小程序即使保存过成功结果，再次提交也让 4G 后端校验原 token，防止旧回执关联到后来的其他绑定。旧 recoveryOnly 入口提示返回 Banban 确认。 + +关联失败不回滚 Banban 绑定。管理页刷新和打开查询/充值入口都会检查原关系，4G 侧独立解绑后不会继续把旧缓存当作有效关联。 + +## 6. 联合解绑流程 + +1. 用户确认后创建持久化 `action=unbind` 操作；此记录阻止同实例再申请 handoff。 +2. 用户点击继续，调用 `POST /internal/v1/binding-relations/{sourceBindingId}/close`，业务请求号为 `bb-close-{operation_id}`。 +3. 返回 `canUnbind=true`：无需跳转，直接执行本地解绑。此判断仅依赖原关系，ICCID 缺失也不会挡住已经得到的服务端确认。 +4. 返回 `bound`：保留 Banban 绑定，携带原 ICCID 跳转 4G 小程序，用户用自己的 4G 登录态调用现有 unbind 接口。 +5. 用户返回、重新打开页面或在 4G 其他入口完成解绑后，可以主动查询/继续处理；最终提交前再次调用幂等 close，只有确认放行才执行本地事务。 +6. 本地设备解绑、家庭关联停用、操作完成在同一事务内提交。失败保留已确认状态，重试仍检查原关系；已完成操作可幂等返回。 +7. 未知关系、鉴权错误、限流、网络异常均保留绑定，不执行自动猜测或无限重试。 + +| 4G 状态 | 处理 | +| --- | --- | +| 原绑定仍有效 | 保留 Banban，前往 4G 解绑 | +| 原绑定已解除 | 关闭后确认 canUnbind=true，再解绑 Banban | +| 有记录证明从未完成绑定 | 撤销待消费凭证并确认放行 | +| 没有关系记录或历史不明确 | unknown，不自动放行，需核查 | + +关闭不会自动解除 4G 用户绑定或删除套餐。它会持续禁止该 sourceBindingId 签发/消费 handoff，文档没有取消关闭接口。页面明确提示继续后不能恢复本次关联；“稍后处理”只是离开页面，不恢复已关闭关系。 + +4G 设备详情页对本机留有 Banban 标记的设备引导回 Banban；清缓存/换手机无法依靠该本地标记拦截所有独立解绑，服务端关系查询负责后续确认。 + +## 7. 有限等待与并发 + +| 环节 | 上限 | +| --- | --- | +| 006 保存 ICCID | 2 秒，失败继续 NFC 绑定 | +| Banban 新业务接口 | 8 秒总期限 | +| Banban 调用 4G | 默认 6 秒网络超时，外层受 8 秒期限约束 | +| Banban 小程序 HTTP | 10 秒，平台超时及独立计时器 | +| 4G 微信登录 | 10 秒 | +| 4G 每次 HTTP | 15 秒，超时终止请求 | +| 4G 单次操作 | 45 秒 | +| 页面及跨小程序跳转 | 15 秒 | + +无自动无限循环或长期轮询，失败后用户主动重试。关闭按 sourceBindingId 幂等；超时不换绑定实例或业务请求号。 + +Banban 不跨网络持有数据库行锁：先读取、关闭事务，再请求 4G，返回后重新校验当前设备所有者和绑定实例。最终事务再次校验操作。4G 端关闭与签发/消费的互斥依赖新版后端的关系锁；本地模拟测试不能替代这部分 MySQL 并发验收。 + +## 8. 配置与数据库 + +```dotenv +TRAFFIC_4G_API_BASE_URL=https://role.api.talkingq.com/traffic-4g +BANBAN_HANDOFF_SERVICE_TOKEN=<两侧一致的服务端密钥> +TRAFFIC_4G_TARGET_APP_ID=wx56e1cfc52aab630f +TRAFFIC_4G_TIMEOUT_SECONDS=6 +``` + +查询和关闭沿用现有密钥。必须先部署 4G 新 JAR、V13 迁移及 Nginx 新路由；仅替换 JAR 不保证公网接口可用。历史关系无法准确回填时保持 unknown。 + +Banban 首次安装本功能：已有库执行一次 `talkingq-url/mysql/migrations/20260923_joint_binding.sql`；新库使用 `mysql/init/02-init.sql` 与 `03-joint-binding.sql`。已执行迁移的生产库不要重跑 ADD COLUMN。本次查询/关闭接入不需要新增 Banban 数据库迁移。 + +## 9. 本地验证与发布 + +```bash +# talkingq-url（既有告警测试断言此测试地址，不需要真实 .env） +SERVER_HOST=192.168.101.78 .venv/bin/python -m pytest tests -q +# banban-mini +npm run test:ui +npm run build:weapp +# 4g-miniprogram +node --test tests/*.test.cjs +``` + +覆盖：原关系解绑、从未绑定、未知关系、旧客户端成功回执、原关系与后来 ICCID 绑定区别、关闭失败重试、事务回滚、重绑后的旧响应、格式/鉴权/限流/超时、页面恢复及跳转计时。 + +状态事务测试使用 SQLite 适配，HTTP 使用模拟传输；没有访问生产 4G 接口或验证真实 MySQL 锁竞争。微信登录、真机跨小程序返回、V13 历史回填和真实关闭/消费并发需部署到联调环境后验收。伴伴构建可执行，全量 TypeScript 检查有既有类型错误，不将构建通过等同于全量类型检查通过。 + +2026-09-23 本次提交内容在独立目录验证：后端 142 项通过，Banban 小程序 30 项通过且微信小程序构建成功，4G 小程序 16 项通过且全部页面脚本/JSON 校验通过。全量 TypeScript 检查的 6 项错误与修改前 HEAD 一致。本次只做本地实施、测试与 Git 交付，不部署新版服务或上传小程序。 + +## 10. 2026-09-23 服务器更新记录 + +按用户授权，于北京时间 14:21 完成 Banban 后端更新，范围为本功能相关的 8 个 Python 文件和已有库迁移脚本。未上传两个小程序，未配置 4G 服务密钥。 + +服务器：`101.35.224.118`，应用目录：`/home/ubuntu/banban-backend`。 + +备份目录：`/home/ubuntu/banban-backups/joint-binding-20260923-1411`，仅 root 可读： + +- `code-and-config.tar.gz`：更新前代码与配置，排除虚拟环境、资源、运行数据和日志。 +- `binding-tables.sql.gz`：`device_settings`、`device_bindings`、`device_family_members`、`device_bind_history` 的一致性快照。 +- `replaced-files/`、`manifest-before.json`、`manifest-after.json`：被替换文件和更新前后 SHA-256 清单。 + +迁移后四表记录数仍为 63、22、33、144；22 条绑定均生成互不重复的绑定实例 UUID。新操作表为空，未为了验证而创建或解除真实绑定。 + +更新后验证：进程 `511137` 启动成功，MQTT 重新连接；7 个新接口路径已注册；设备所有者状态查询与原绑定查询均为 HTTP 200；未关联设备的查询/充值跳转接口返回预期 409；未登录访问返回 401；公网 HTTPS OpenAPI 和绑定提示音均为 200。 + +截至 14:22，`TalkingQ_ZHWK00003111` 的 ICCID 仍为空，尚无更新后的 `006 ICCID` 检测日志，需要设备重新上报。`BANBAN_HANDOFF_SERVICE_TOKEN` 仍未配置，申请 4G 凭证暂不可用。 + +14:25 按用户提供的密钥配置 `BANBAN_HANDOFF_SERVICE_TOKEN`。更新前 `.env` 备份于 `/home/ubuntu/banban-backups/handoff-token-20260923-1424/env.before`；其他配置项保持原值，现有 `.env` 权限收紧为 600。独立进程校验有效配置与提供值一致，重启后进程为 `512483`，HTTPS OpenAPI 为 200，MQTT 连接正常。密钥值不记录到文档或仓库。 + +4G 申请接口空请求分别在无密钥、有密钥时均返回 400 `INVALID_REQUEST`，说明该检查无法证明密钥鉴权成功；未提交有效 ICCID 或创建交接凭证，实际关联仍待正常业务请求验证。 diff --git a/docs/设备直接绑定接口联调文档(2).md b/docs/设备直接绑定接口联调文档(2).md new file mode 100644 index 0000000..621d613 --- /dev/null +++ b/docs/设备直接绑定接口联调文档(2).md @@ -0,0 +1,564 @@ +# 设备直接绑定接口联调文档 + +## 1. 适用范围 + +本文档用于 Banban 小程序与 4G 管理后端之间的“设备无法获取绑定码时直接绑定”联调。 + +2026-09-23 更新：新增按 `sourceBindingId` 查询原绑定关系、关闭 handoff 关联的服务端接口，见第 11、12 节。伴伴解绑必须以 `canUnbind=true` 为放行条件，不能只看 ICCID 当前状态或把查不到当成已经解绑。 + +生产环境推荐使用 `handoffToken` 两段式流程： + +1. Banban 后端使用服务凭证，向 4G 管理后端申请一次性 `handoffToken`。 +2. 4G 小程序用户完成微信登录后，携带程序用户 Token 和 `handoffToken` 完成设备绑定。 + +该流程仍然复用普通绑定码绑定的事务逻辑，成功后会： + +- 创建当前 4G 小程序用户与设备的有效绑定关系； +- 将 `sim_card.binding_status` 更新为 `bound`； +- 作废该设备尚未使用的绑定验证码； +- 清理绑定验证码缓存； +- 写入绑定审计记录。 + +handoff 申请接口会在服务 Token 校验通过后，按 ICCID 幂等创建或复用 4G 后端的 `sim_card` 档案。新建档案初始状态为 `active`、`unbound`、`sync_status=pending`；消费 handoffToken 后才会创建用户绑定并将绑定状态更新为 `bound`。因此 Banban 不需要提前调用 4G 后端的设备注册接口。 + +该自动建档能力只对已认证的 Banban 服务接口开放，不代表可以通过公开接口任意写入 ICCID。ICCID 格式仍必须合法；如果已有档案状态为非 `active`，申请会被拒绝。 + +## 2. 联调环境与配置 + +当前生产环境通过 Nginx 的 `/traffic-4g` 前缀访问，以下 `BASE_URL` 必须包含此前缀： + +```text +BASE_URL=https://role.api.talkingq.com/traffic-4g +``` + +4G 后端需要配置并重启后才会生效： + +```bash +HANDOFF_ENABLED=true +HANDOFF_TOKEN_TTL=5m +HANDOFF_TOKEN_PEPPER=<独立的高强度随机值> +BANBAN_HANDOFF_SERVICE_TOKEN=<独立的高强度随机值> +WECHAT_APP_ID=<4G 小程序 AppID> +DIRECT_ICCID_BINDING_ENABLED=false +``` + +说明： + +- `BANBAN_HANDOFF_SERVICE_TOKEN` 只允许保存在 Banban 后端服务端，不能下发到小程序或前端页面。 +- `HANDOFF_TOKEN_PEPPER` 只允许保存在 4G 后端服务端，不能与服务 Token 共用。 +- `HANDOFF_TOKEN_TTL=5m` 表示凭证默认有效 5 分钟，具体有效期以申请接口返回的 `expiresIn` 为准。 +- 当前环境的 `BINDING_ALLOWED_ORIGINS` 可以继续保留 `https://banban.api.talkingq.com,http://192.168.101.151`，但 handoff 流程不依赖 Origin 证明设备归属。 +- 新增关系查询、关闭接口继续使用 `BANBAN_HANDOFF_SERVICE_TOKEN`，不需要新增密钥或修改 `.env`。即使 `HANDOFF_ENABLED=false`，仍可使用正确服务凭证查询、关闭已有关系。 + +## 3. 完整调用时序 + +```text +Banban 后端 4G 管理后端 4G 小程序 + | | | + |-- X-Service-Token + ICCID -------->| | + |<--------- handoffToken ------------| | + | | | + | |<-- wx.login code --------| + | |---- accessToken -------->| + | | | + | |<-- accessToken + token --| + | |---- 完成设备绑定 -------->| +``` + +Banban 后端申请凭证的接口只能由服务端调用；4G 小程序消费凭证的接口只能由登录后的 `PROGRAM_USER` 调用。 + +## 4. 接口一：Banban 后端申请一次性 handoffToken + +### 4.1 请求地址 + +```http +POST /internal/v1/handoff-tokens +``` + +完整地址： + +```text +https://role.api.talkingq.com/traffic-4g/internal/v1/handoff-tokens +``` + +必须使用 HTTPS。该接口不使用小程序 JWT，而是使用服务间 Token。 + +### 4.2 请求头 + +```http +X-Service-Token: +Content-Type: application/json +X-Request-Id: banban-handoff-20260922-0001 +``` + +`X-Service-Token` 的值必须与 4G 后端环境变量 `BANBAN_HANDOFF_SERVICE_TOKEN` 完全一致。 + +`X-Request-Id` 仅用于 HTTP 请求追踪，建议使用 8 至 64 位的字母、数字、点、下划线或短横线。真正的业务幂等键是请求体中的 `requestId`。 + +### 4.3 请求体 + +```json +{ + "iccid": "898604E91123D1334901", + "sourceBindingId": "banban-binding-100001", + "requestId": "banban-handoff-20260922-0001" +} +``` + +| 字段 | 类型 | 必填 | 约束与说明 | +| --- | --- | --- | --- | +| `iccid` | string | 是 | 18 至 22 位字母或数字，后端会统一转为大写；已登记的 active 档案会复用，未登记时由 4G 后端自动建档 | +| `sourceBindingId` | string | 是 | Banban 侧绑定关系的稳定唯一 ID，最长 128 个字符；同一绑定关系保持不变，且必须始终对应同一 ICCID。新接入使用 UUID 或字母、数字、短横线、下划线，便于作为查询路径参数 | +| `requestId` | string | 是 | 本次申请的幂等请求号，最长 64 个字符；网络超时重试必须保持不变 | + +`sourceBindingId` 不要使用每次随机生成的值，否则无法实现申请接口幂等。`requestId` 建议由 Banban 后端生成，例如 `banban-handoff-20260922-0001`。 + +即使换了 `requestId`，同一 `sourceBindingId` 也不能切换 ICCID，否则返回 `IDEMPOTENCY_CONFLICT`。伴伴关系进入 `closing` 后不得再申请或消费 handoff；完成解绑并重新建立伴伴绑定时，必须使用新关系的唯一 ID，不能复用旧 ID 绕过关闭状态。 + +### 4.4 curl 示例 + +以下命令只能在 Banban 后端服务器或安全的服务端环境执行： + +```bash +export BASE_URL="https://role.api.talkingq.com/traffic-4g" +export BANBAN_HANDOFF_SERVICE_TOKEN="<服务端密钥>" + +curl -i -sS -X POST "$BASE_URL/internal/v1/handoff-tokens" \ + -H "X-Service-Token: $BANBAN_HANDOFF_SERVICE_TOKEN" \ + -H "Content-Type: application/json" \ + -H "X-Request-Id: banban-handoff-20260922-0001" \ + --data '{ + "iccid": "898604E91123D1334901", + "sourceBindingId": "banban-binding-100001", + "requestId": "banban-handoff-20260922-0001" + }' +``` + +### 4.5 成功响应 + +HTTP 状态码为 `200`： + +```json +{ + "code": "OK", + "message": "success", + "data": { + "targetAppId": "<4G 小程序 AppID>", + "handoffToken": "<一次性凭证>", + "expiresIn": 300 + }, + "requestId": "banban-handoff-20260922-0001", + "timestamp": "2026-09-22T03:30:00Z" +} +``` + +| 字段 | 说明 | +| --- | --- | +| `data.targetAppId` | 允许消费该凭证的目标小程序 AppID，必须与 4G 小程序登录身份一致 | +| `data.handoffToken` | 一次性绑定凭证，只能传给 4G 小程序的绑定接口，不要写入日志或返回给不可信客户端 | +| `data.expiresIn` | 剩余有效秒数，消费时以服务端校验结果为准 | + +相同的 `sourceBindingId + requestId` 重试时： + +- 凭证仍为 `active` 且未过期：返回同一凭证，不会重复创建； +- 对应 ICCID 发生变化：返回 `IDEMPOTENCY_CONFLICT`； +- 凭证已消费：返回 `HANDOFF_TOKEN_USED`； +- 凭证已过期：返回 `HANDOFF_TOKEN_EXPIRED`。 + +## 5. 接口二：4G 小程序消费 handoffToken 完成绑定 + +### 5.1 先获取 4G 小程序 accessToken + +4G 小程序先调用微信 `wx.login()`，再把返回的临时 `code` 交给 4G 后端： + +```http +POST /api/v1/program/auth/login +``` + +请求： + +```http +Content-Type: application/json +X-Request-Id: program-login-20260922-0001 +``` + +```json +{ + "code": "<4G 小程序 wx.login 返回的 code>" +} +``` + +curl 示例： + +```bash +curl -i -sS -X POST "$BASE_URL/api/v1/program/auth/login" \ + -H "Content-Type: application/json" \ + -H "X-Request-Id: program-login-20260922-0001" \ + --data '{"code":"<4G 小程序 wx.login 返回的 code>"}' +``` + +成功响应中的 `data.accessToken` 是后续绑定接口需要的程序用户 Token： + +```json +{ + "code": "OK", + "message": "success", + "data": { + "accessToken": "", + "expiresAt": "2026-09-22T15:30:00Z", + "user": { + "userId": 10001, + "status": "active" + } + }, + "requestId": "program-login-20260922-0001", + "timestamp": "2026-09-22T03:31:00Z" +} +``` + +必须使用 4G 小程序自己的 `wx.login()` code，不能使用 Banban 小程序的登录 code，也不能使用 `SUPER_ADMIN`、`FACTORY_PROVISIONER` 或设备 Token。 + +### 5.2 请求地址 + +```http +POST /api/v1/program/bindings/handoff +``` + +完整地址： + +```text +https://role.api.talkingq.com/traffic-4g/api/v1/program/bindings/handoff +``` + +### 5.3 请求头 + +```http +Authorization: Bearer +Content-Type: application/json +X-Request-Id: program-handoff-20260922-0001 +``` + +该接口不要求 `Origin` 白名单。权限来自程序用户 JWT、目标小程序 AppID 校验和一次性 `handoffToken`。 + +### 5.4 请求体 + +```json +{ + "handoffToken": "<接口一返回的一次性凭证>", + "requestId": "program-handoff-20260922-0001" +} +``` + +| 字段 | 类型 | 必填 | 约束与说明 | +| --- | --- | --- | --- | +| `handoffToken` | string | 是 | 接口一返回的一次性凭证，最长 256 个字符；不要自行修改或解码 | +| `requestId` | string | 是 | 本次绑定的幂等请求号，最长 64 个字符；网络超时重试时保持不变 | + +### 5.5 curl 示例 + +```bash +export PROGRAM_ACCESS_TOKEN="<4G 小程序登录返回的 accessToken>" +export HANDOFF_TOKEN="<接口一返回的 handoffToken>" + +curl -i -sS -X POST "$BASE_URL/api/v1/program/bindings/handoff" \ + -H "Authorization: Bearer $PROGRAM_ACCESS_TOKEN" \ + -H "Content-Type: application/json" \ + -H "X-Request-Id: program-handoff-20260922-0001" \ + --data "{\"handoffToken\":\"$HANDOFF_TOKEN\",\"requestId\":\"program-handoff-20260922-0001\"}" +``` + +### 5.6 成功响应 + +HTTP 状态码为 `200`： + +```json +{ + "code": "OK", + "message": "success", + "data": { + "iccid": "898604E91123D1334901", + "alreadyBound": false + }, + "requestId": "program-handoff-20260922-0001", + "timestamp": "2026-09-22T03:32:00Z" +} +``` + +| 字段 | 说明 | +| --- | --- | +| `data.iccid` | 实际绑定成功的设备 ICCID | +| `data.alreadyBound` | `false` 表示本次新建绑定；`true` 表示重复请求返回已有绑定结果 | + +绑定、作废旧验证码、更新 SIM 绑定状态、记录 handoff 对应的具体绑定记录、消费凭证和审计记录在同一个数据库事务中完成。即使设备原本已绑定当前用户，也会记录复用的绑定记录 ID。接口成功后，可以调用 `GET /api/v1/program/devices` 验证当前用户的设备列表。 + +## 6. 幂等与重试规则 + +### 6.1 申请凭证重试 + +Banban 后端申请接口超时或未收到响应时，使用相同的 `sourceBindingId` 和 `requestId` 重试，不要重新生成请求号。这样可以复用原有有效凭证，避免创建多条 handoff 记录。 + +### 6.2 消费凭证重试 + +4G 小程序消费接口超时或网络断开时，使用相同的 `handoffToken` 和 `requestId` 重试： + +- 如果事务已经成功提交，原绑定仍有效且伴伴关系未关闭，原消费用户重试返回 `alreadyBound=true`； +- 如果凭证已被其他用户消费，或对应的原绑定已解绑，返回 `HANDOFF_TOKEN_USED`；设备后来重新绑定给同一用户，也不能复用旧凭证； +- 如果伴伴已关闭此关系，返回 `HANDOFF_RELATION_CLOSED`； +- 不要因为超时就重新向 Banban 申请新的 `sourceBindingId`，应先使用原请求参数重试或查询绑定结果。 + +### 6.3 凭证安全规则 + +- `handoffToken` 只能在服务端申请、在 4G 小程序绑定流程中消费一次； +- 不要把 `BANBAN_HANDOFF_SERVICE_TOKEN` 放在小程序、浏览器代码、页面参数或日志中； +- 不要把完整 `handoffToken` 写入业务日志、埋点或群聊； +- `handoffToken` 过期、篡改或消费后不可恢复，需按原绑定关系重新申请。 + +## 7. 错误码与处理建议 + +统一错误响应结构如下，`data` 为 `null`： + +```json +{ + "code": "HANDOFF_TOKEN_EXPIRED", + "message": "跨系统绑定凭证已过期", + "data": null, + "requestId": "program-handoff-20260922-0001", + "timestamp": "2026-09-22T03:40:00Z" +} +``` + +### 7.1 申请凭证接口 + +| HTTP 状态 | `code` | 含义与处理 | +| --- | --- | --- | +| 400 | `INVALID_REQUEST` | ICCID、`sourceBindingId` 或 `requestId` 格式不符合约束 | +| 401 | `HANDOFF_SERVICE_UNAUTHORIZED` | 缺少或错误的 `X-Service-Token`；确认服务端密钥配置一致 | +| 403 | `DEVICE_DISABLED` | 设备状态不是 `active`；先恢复设备状态 | +| 409 | `IDEMPOTENCY_CONFLICT` | 相同幂等键对应了不同 ICCID；检查 `sourceBindingId` 和 `requestId` | +| 409 | `HANDOFF_TOKEN_USED` | 原凭证已经消费；按当前绑定结果处理或重新发起新业务请求 | +| 409 | `HANDOFF_RELATION_CLOSED` | 此伴伴关系已停止新关联；不能重新申请凭证 | +| 409 | `HANDOFF_RELATION_UNKNOWN` | 历史记录无法准确关联，需要核查，不能绕过检查重新申请 | +| 410 | `HANDOFF_TOKEN_EXPIRED` | 原凭证已过期；使用相同业务关系重新申请凭证 | +| 429 | `RATE_LIMITED` | 触发申请频率限制；降低重试频率 | +| 503 | `HANDOFF_DISABLED` | 后端未启用 handoff 流程；检查 `HANDOFF_ENABLED` | +| 503 | `CONFIGURATION_ERROR` | handoff 密钥、4G AppID 或 TTL 配置缺失/不合法 | +| 500 | `INTERNAL_ERROR` | 自动建档后无法读取 SIM 档案；需要联系后端排查数据库或事务状态 | + +### 7.2 消费凭证接口 + +| HTTP 状态 | `code` | 含义与处理 | +| --- | --- | --- | +| 400 | `INVALID_REQUEST` | 请求体字段缺失或超过长度限制 | +| 400 | `HANDOFF_TOKEN_INVALID` | 凭证不存在、被篡改或已失效 | +| 401 | `UNAUTHORIZED` | 未登录、Token 过期或 Token 不是程序用户 Token | +| 403 | `HANDOFF_TARGET_APP_MISMATCH` | 当前登录的小程序不是接口一返回的 `targetAppId` | +| 403 | `DEVICE_DISABLED` | 设备当前被禁用 | +| 409 | `HANDOFF_TOKEN_USED` | 凭证已使用且无法匹配当前用户的重试 | +| 409 | `DEVICE_ALREADY_BOUND` | 设备已经绑定其他程序用户 | +| 409 | `IDEMPOTENCY_CONFLICT` | 相同 `requestId` 对应了不同设备或不同绑定操作 | +| 409 | `HANDOFF_RELATION_CLOSED` | 伴伴关系已关闭；已有未消费凭证不能再绑定 | +| 409 | `HANDOFF_RELATION_UNKNOWN` | 无法确认此凭证对应的原关系，需联系后端核查 | +| 410 | `HANDOFF_TOKEN_EXPIRED` | 凭证已经过期 | +| 429 | `RATE_LIMITED` | 用户、IP 或凭证消费频率过高 | +| 503 | `HANDOFF_DISABLED` | 后端未启用 handoff 流程 | + +## 8. 兼容接口：裸 ICCID 直接绑定（逐步下线） + +旧接口地址： + +```http +POST /api/v1/program/bindings/by-iccid +``` + +该接口要求： + +- 使用 `PROGRAM_USER` 的 `Authorization: Bearer` Token； +- 携带与页面完全一致的 `Origin`； +- `Origin` 必须存在于 `BINDING_ALLOWED_ORIGINS`； +- 后端配置 `DIRECT_ICCID_BINDING_ENABLED=true`。 + +当前环境已配置： + +```bash +DIRECT_ICCID_BINDING_ENABLED=false +``` + +因此当前调用该接口会返回 HTTP `410` 和 `DIRECT_BINDING_DISABLED`。只有临时兼容旧版本联调时，才允许在服务端显式开启： + +```bash +DIRECT_ICCID_BINDING_ENABLED=true +BINDING_ALLOWED_ORIGINS=https://banban.api.talkingq.com,http://192.168.101.151 +``` + +旧接口请求示例： + +```http +POST /api/v1/program/bindings/by-iccid +Origin: https://banban.api.talkingq.com +Authorization: Bearer +Content-Type: application/json +X-Request-Id: direct-bind-20260922-0001 +``` + +```json +{ + "iccid": "898604E91123D1334901", + "requestId": "direct-bind-20260922-0001" +} +``` + +该接口无法证明 Banban 后端实际拥有该 ICCID，不应作为新的跨系统正式方案。新接入方必须使用本文档第 4、5 节的 handoffToken 流程。 + +## 9. 联调验收清单 + +建议按以下顺序验收： + +1. 使用尚未登记的测试 ICCID，申请接口返回 `code=OK`、`targetAppId`、`handoffToken` 和正数 `expiresIn`。 +2. 查询数据库，确认该 ICCID 已自动写入 `sim_card`，初始为 `status=active`、`binding_status=unbound`、`sync_status=pending`。 +3. 使用 4G 小程序的 `wx.login()` code 登录，确认返回 `PROGRAM_USER` 的 `accessToken`。 +4. 使用该 `accessToken` 消费 handoffToken，返回 `code=OK`、正确 ICCID、`alreadyBound=false`。 +5. 查询 `GET /api/v1/program/devices`，确认设备出现在当前 4G 小程序用户列表中。 +6. 使用相同 `sourceBindingId + requestId` 重试申请接口，确认不会重复创建 SIM 档案或凭证。 +7. 使用相同用户、相同 handoffToken 和相同绑定 `requestId` 重试消费接口，确认返回幂等结果或可确认已绑定。 +8. 绑定成功后调用流量详情接口，确认设备可以进入 `pending` 或同步后的流量状态，而不是 `DEVICE_NOT_FOUND`。 +9. 使用其他小程序身份消费，确认返回 `HANDOFF_TARGET_APP_MISMATCH`。 +10. 使用过期或篡改后的凭证，确认分别返回 `HANDOFF_TOKEN_EXPIRED` 或 `HANDOFF_TOKEN_INVALID`。 +11. 使用已绑定到其他用户的设备，确认返回 `DEVICE_ALREADY_BOUND`，不会覆盖原绑定关系。 +12. 确认数据库中该设备的 `sim_card.binding_status` 为 `bound`，且 `user_device_binding.status` 为 `active`。 +13. 确认旧 `/bindings/by-iccid` 接口在 `DIRECT_ICCID_BINDING_ENABLED=false` 时返回 `DIRECT_BINDING_DISABLED`。 + +## 10. 联调排查信息 + +请保留以下信息用于后端排查，但不要提交 Token 或完整 ICCID： + +- 请求路径和 HTTP 状态码； +- 响应中的 `code`、`message`、`requestId`； +- 请求头中的 `X-Request-Id`； +- Banban 的 `sourceBindingId` 和脱敏后的 ICCID； +- 调用时间（含时区）； +- 是否使用了相同幂等参数重试。 + +不要在工单、群聊、代码仓库或日志中粘贴 `X-Service-Token`、`PROGRAM_USER_ACCESS_TOKEN` 或完整 `handoffToken`。 + +## 11. 接口三：查询这次伴伴绑定对应的 4G 关系 + +```http +GET /internal/v1/binding-relations/{sourceBindingId} +X-Service-Token: +``` + +查询的是 `sourceBindingId` 实际关联的具体 4G 绑定记录。原绑定解绑后，即使同一 ICCID 后来被其他用户或原用户通过其他流程重新绑定，原关系仍返回 `unbound`。若未关闭的同一伴伴关系又成功消费了新 handoff，则查询追踪这次新关联。 + +```bash +export BASE_URL="https://role.api.talkingq.com/traffic-4g" +export SOURCE_BINDING_ID="banban-binding-100001" + +curl -i -sS "$BASE_URL/internal/v1/binding-relations/$SOURCE_BINDING_ID" \ + -H "X-Service-Token: $BANBAN_HANDOFF_SERVICE_TOKEN" \ + -H "X-Request-Id: relation-query-20260923-0001" +``` + +成功响应示例（四种业务状态均返回 HTTP 200）： + +```json +{ + "code": "OK", + "message": "success", + "data": { + "sourceBindingId": "banban-binding-100001", + "status": "never_bound", + "pendingHandoff": true, + "acceptingHandoff": true, + "relationState": "open", + "canUnbind": false, + "reason": "NO_COMPLETED_BINDING", + "checkedAt": "2026-09-23T08:00:00Z" + }, + "requestId": "relation-query-20260923-0001", + "timestamp": "2026-09-23T08:00:00Z" +} +``` + +| `status` | 含义 | 伴伴处理 | +| --- | --- | --- | +| `bound` | 本次伴伴关系关联的 4G 绑定仍有效 | 提示先完成 4G 解绑 | +| `unbound` | 关联的原 4G 绑定已由用户或管理员解除 | 完成关闭后，仅 `canUnbind=true` 才能解绑伴伴 | +| `never_bound` | 有关系记录，且能证明尚未完成过 handoff 绑定 | 完成关闭后，仅 `canUnbind=true` 才能解绑伴伴 | +| `unknown` | 无记录、历史无法准确关联或数据不一致 | 提示未确认，保留伴伴绑定并核查 | + +| 字段 | 说明 | +| --- | --- | +| `pendingHandoff` | 是否存在 `active` 且未过期的凭证；数据库状态尚未改为 `expired` 的过期凭证不计入 | +| `acceptingHandoff` | 此关系是否仍允许申请或消费新 handoff；还需满足功能开关、身份、设备状态等原有检查 | +| `relationState` | `open` 允许关联；`closing` 已停止关联；`unknown` 表示没有关系记录 | +| `canUnbind` | 仅当关系已 `closing`、无待处理凭证，且 `status` 为 `unbound/never_bound` 时为 `true` | +| `reason` | 状态原因，例如 `NO_RELATION`、`AMBIGUOUS_HISTORY`、`ORIGINAL_BINDING_ACTIVE`、`ORIGINAL_BINDING_RELEASED`、`NO_COMPLETED_BINDING` | +| `checkedAt` | 查询时间，UTC、带 `Z` 时区；只是本次快照时间，不是后续请求的锁或授权凭据 | + +该 GET 不会撤销凭证或关闭关系，响应禁止缓存。即使查到 `never_bound + pendingHandoff=false`，只要仍为 `open`，`canUnbind` 就是 `false`。伴伴不能根据一次普通 GET 直接删除绑定，必须进入下一节关闭流程。 + +## 12. 接口四：关闭关联并确认是否可以解绑伴伴 + +```http +POST /internal/v1/binding-relations/{sourceBindingId}/close +X-Service-Token: +Content-Type: application/json + +{"requestId":"banban-close-20260923-0001"} +``` + +`requestId` 必填，最长 64 字符，用于审计及重试追踪；关闭操作按 `sourceBindingId` 幂等，重复调用不会恢复关系，也不会重复解绑设备。返回结构与第 11 节一致。 + +```bash +curl -i -sS -X POST "$BASE_URL/internal/v1/binding-relations/$SOURCE_BINDING_ID/close" \ + -H "X-Service-Token: $BANBAN_HANDOFF_SERVICE_TOKEN" \ + -H "Content-Type: application/json" \ + -H "X-Request-Id: banban-close-20260923-0001" \ + --data '{"requestId":"banban-close-20260923-0001"}' +``` + +此操作会锁定关系，将其设置为 `closing`，撤销所有尚未消费的凭证，再判断原绑定是否有效。不会自动解除 4G 用户绑定，也不会删除 SIM 档案或影响已有流量、充值记录。 + +如果 `sourceBindingId` 完全没有记录，关闭操作会创建一条拒绝后续关联的记录，但仍返回 `unknown + canUnbind=false`，不会把“没有记录”变成“证明从未绑定”。 + +伴伴后端的解绑顺序： + +1. 用户确认解绑后，先将本地伴伴关系标记为解绑中；该状态必须持久化，并阻止所有入口发起新的 handoff。 +2. 调用上述 `close`；超时、网络失败时保持解绑中并重试，不删除本地绑定。 +3. 返回 `bound`：提示用户先到 4G 小程序解绑。此时 handoff 已被关闭，但原 4G 绑定继续有效，流量查询和充值仍按原权限工作。 +4. 用户完成 4G 解绑后，再调用 GET 或重试 `close`；只有 `canUnbind=true` 时才能删除伴伴绑定。 +5. 返回 `unknown`，或 HTTP 401/429/5xx：保留伴伴绑定，提示未确认或稍后重试。不能把异常兜底成 `unbound`。 + +并发结果：消费事务先获得关系锁并完成绑定时，`close` 等待后返回 `bound`；`close` 先获得锁时，正在途中的签发、消费请求等待后返回 `HANDOFF_RELATION_CLOSED`。`pendingHandoff` 是已提交凭证的快照；正在进行的消费由 `close` 的事务锁处理。 + +`closing` 是持续生效的关闭状态，本版本没有恢复/取消关闭接口。即使用户取消了伴伴解绑，也不会自动恢复此 ID 的 handoff 能力；已有 4G 绑定可以继续使用。只有伴伴真正建立新的绑定关系时才使用新的 `sourceBindingId`。 + +查询、关闭接口错误码： + +| HTTP 状态 | `code` | 处理 | +| --- | --- | --- | +| 400 | `INVALID_REQUEST` | 路径 ID 为空/超长，或关闭请求缺少合法 `requestId` | +| 401 | `HANDOFF_SERVICE_UNAUTHORIZED` | 缺失或错误的服务凭证 | +| 429 | `RATE_LIMITED` | 同一关系查询和关闭合计超过每分钟 60 次；降低频率 | +| 500 | `INTERNAL_ERROR` | 数据库、事务或其他内部失败；保留伴伴关系并重试/排查 | + +## 13. 本次升级部署及验收 + +本次新增 Flyway `V13__add_handoff_binding_relations.sql`，启动时自动建表、关联凭证并回填可明确识别的历史绑定，不修改已有 V12 迁移文件。历史凭证能通过消费用户、SIM、消费请求号唯一关联原绑定时会回填；同用户复用旧绑定但请求号不同等无法确认的情况保留为 `unknown`，由后端核查。 + +发布时同步更新 JAR 与 `nginx/locations.conf`，让新增 `/traffic-4g/internal/v1/binding-relations/` 路径生效；在实际承载域名的 Nginx 进程/容器中检查配置并重载。只替换 JAR 不会使 Nginx 自动开放新路径。迁移期间停止旧后端实例写入，完成迁移后再由新实例提供 handoff 服务。 + +新增验收场景： + +1. 不带或带错误 `X-Service-Token` 调用查询、关闭，均返回 HTTP 401。 +2. 查询未出现过的 `sourceBindingId`，返回 HTTP 200、`unknown`、`canUnbind=false`。 +3. 申请未消费凭证后，查询返回 `never_bound`、`pendingHandoff=true`、`canUnbind=false`。 +4. 同一关系申请多张凭证后关闭，全部不能再消费，返回 `never_bound`、`pendingHandoff=false`、`canUnbind=true`。 +5. 成功消费后关闭，返回 `bound`、`canUnbind=false`；用户或管理员完成 4G 解绑后，查询返回 `unbound`、`canUnbind=true`。 +6. 原绑定解绑后，同一 ICCID 被其他用户或原用户通过其他流程绑定，旧关系查询仍为 `unbound`；旧已消费凭证不能变成新绑定的重试凭证。 +7. 同一 `sourceBindingId` 使用不同 `requestId` 换 ICCID，返回 `IDEMPOTENCY_CONFLICT`。 +8. 并发执行关闭与签发/消费，结果只能是“先绑定成功，关闭返回 bound”或“关闭成功，后续关联失败”。 +9. 重复关闭结果保持关闭状态；没有历史的关系关闭后仍为 `unknown`。 +10. 历史不明确的数据保持 `unknown`，不得通过换申请请求号绕过。 + +自动化验证：普通单元和接口测试执行 `mvn test`；真实 MySQL 事务测试为 `HandoffBindingRelationIntegrationTest`，需要显式配置 `HANDOFF_TEST_JDBC_URL`、`HANDOFF_TEST_DB_PASSWORD`（可选 `HANDOFF_TEST_DB_USERNAME`，默认 root），指向全新的空测试库后执行 `mvn -Dtest=HandoffBindingRelationIntegrationTest test`。该测试只使用指定测试库，不读取生产 `.env`；数据库非空时会拒绝执行。 diff --git a/talkingq-url/banban/dao/binding.py b/talkingq-url/banban/dao/binding.py index 178b4cf..abdc907 100644 --- a/talkingq-url/banban/dao/binding.py +++ b/talkingq-url/banban/dao/binding.py @@ -108,7 +108,7 @@ class BindingDAO(BaseDAO): async def _bind_device(self, device_id: str, user_id: int, child_id: Optional[int]) -> None: existing_by_device = ( await self.execute( - "SELECT id, status FROM device_bindings WHERE device_id = :device_id", + "SELECT id, status, owner_user_id, binding_instance_id FROM device_bindings WHERE device_id = :device_id FOR UPDATE", {"device_id": device_id}, ) ).mappings().first() @@ -122,18 +122,20 @@ class BindingDAO(BaseDAO): child_id = NULL, status = 1, unbound_at = NULL, + binding_instance_id = :binding_instance_id, updated_at = CURRENT_TIMESTAMP WHERE device_id = :device_id """, - {"owner_user_id": user_id, "device_id": device_id}, + {"owner_user_id": user_id, "device_id": device_id, + "binding_instance_id": self._binding_instance(existing_by_device, user_id)}, ) else: await self.execute( """ - INSERT INTO device_bindings (device_id, owner_user_id, child_id, status, bound_at) - VALUES (:device_id, :owner_user_id, NULL, 1, CURRENT_TIMESTAMP) + INSERT INTO device_bindings (device_id, owner_user_id, child_id, status, bound_at, binding_instance_id) + VALUES (:device_id, :owner_user_id, NULL, 1, CURRENT_TIMESTAMP, :binding_instance_id) """, - {"device_id": device_id, "owner_user_id": user_id}, + {"device_id": device_id, "owner_user_id": user_id, "binding_instance_id": str(uuid.uuid4())}, ) return @@ -156,10 +158,12 @@ class BindingDAO(BaseDAO): child_id = :child_id, status = 1, unbound_at = NULL, + binding_instance_id = :binding_instance_id, updated_at = CURRENT_TIMESTAMP WHERE device_id = :device_id """, - {"owner_user_id": user_id, "child_id": child_id, "device_id": device_id}, + {"owner_user_id": user_id, "child_id": child_id, "device_id": device_id, + "binding_instance_id": self._binding_instance(existing_by_device, user_id)}, ) return @@ -168,12 +172,18 @@ class BindingDAO(BaseDAO): await self.execute( """ - INSERT INTO device_bindings (device_id, owner_user_id, child_id, status, bound_at) - VALUES (:device_id, :owner_user_id, :child_id, 1, CURRENT_TIMESTAMP) + INSERT INTO device_bindings (device_id, owner_user_id, child_id, status, bound_at, binding_instance_id) + VALUES (:device_id, :owner_user_id, :child_id, 1, CURRENT_TIMESTAMP, :binding_instance_id) """, - {"device_id": device_id, "owner_user_id": user_id, "child_id": child_id}, + {"device_id": device_id, "owner_user_id": user_id, "child_id": child_id, "binding_instance_id": str(uuid.uuid4())}, ) + @staticmethod + def _binding_instance(row, user_id): + if int(row["status"]) == 1 and int(row["owner_user_id"]) == user_id and row.get("binding_instance_id"): + return row["binding_instance_id"] + return str(uuid.uuid4()) + async def _ensure_owner_family_member(self, device_id: str, user_id: int) -> None: await self.execute( """ @@ -607,7 +617,7 @@ class BindingDAO(BaseDAO): await self.commit() return True - async def unbind(self, device_id: str, user_id: int) -> bool: + async def unbind(self, device_id: str, user_id: int, *, commit: bool = True) -> bool: row = ( await self.execute( """ @@ -667,7 +677,8 @@ class BindingDAO(BaseDAO): """, {"device_id": device_id, "user_id": user_id}, ) - await self.commit() + if commit: + await self.commit() return True async def list_history(self, device_id: str, limit: int = 20, cursor: datetime = None) -> list[Mapping]: diff --git a/talkingq-url/banban/routers/__init__.py b/talkingq-url/banban/routers/__init__.py index 898365c..b00e298 100644 --- a/talkingq-url/banban/routers/__init__.py +++ b/talkingq-url/banban/routers/__init__.py @@ -1,6 +1,7 @@ from fastapi import APIRouter from banban.routers.bindings import router as bindings_router +from banban.routers.joint_binding import router as joint_binding_router from banban.routers.children import router as children_router from banban.routers.device_im import router as device_im_router from banban.routers.device_location import router as device_location_router @@ -20,6 +21,7 @@ banban_router = APIRouter(prefix="/banban") banban_router.include_router(wechat_auth_router, tags=["banban-auth"]) banban_router.include_router(wechat_mp_router, tags=["banban-wechat-mp"]) banban_router.include_router(bindings_router, tags=["banban-bindings"]) +banban_router.include_router(joint_binding_router) banban_router.include_router(children_router, tags=["banban-children"]) banban_router.include_router(devices_router, tags=["banban-devices"]) banban_router.include_router(family_router, tags=["banban-family"]) diff --git a/talkingq-url/banban/routers/bindings.py b/talkingq-url/banban/routers/bindings.py index c78393f..626123b 100644 --- a/talkingq-url/banban/routers/bindings.py +++ b/talkingq-url/banban/routers/bindings.py @@ -375,8 +375,7 @@ async def unbind_device( current_user_id: int = Depends(get_current_user_id), ) -> None: del request - if not await binding_service.unbind(device_id, current_user_id): - raise HTTPException(status_code=404, detail="binding not found") + raise HTTPException(status_code=409, detail="请使用联合解绑流程，完成流量侧解绑后再解除设备绑定") @router.get("/history/{device_id}", response_model=BindHistoryResponse) diff --git a/talkingq-url/banban/routers/joint_binding.py b/talkingq-url/banban/routers/joint_binding.py new file mode 100644 index 0000000..94fc3ca --- /dev/null +++ b/talkingq-url/banban/routers/joint_binding.py @@ -0,0 +1,74 @@ +import asyncio +from typing import Literal +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, Field, StrictBool + +from banban.security import get_current_user_id +from banban.service.joint_binding import joint_binding_service +from banban.service.traffic_handoff import TrafficError + +router = APIRouter(prefix="/joint-bindings", tags=["joint-bindings"]) + + +class StartRequest(BaseModel): + device_id: str = Field(min_length=1, max_length=64) + action: Literal["bind", "unbind"] + + +class ResultRequest(BaseModel): + action: Literal["bind", "unbind"] + request_id: str = Field(min_length=1, max_length=64) + success: StrictBool + code: str = Field(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_]+$") + iccid: str | None = Field(default=None, max_length=22) + + +async def call(coro): + try: + return await asyncio.wait_for(coro, timeout=8) + except asyncio.TimeoutError: + raise HTTPException(503, detail={"code": "OPERATION_TIMEOUT", "message": "操作暂时无法确认，请重试"}) from None + except TrafficError as exc: + raise HTTPException(exc.status_code, detail={"code": exc.code, "message": str(exc)}) from None + + +@router.post("") +async def start(payload: StartRequest, user: int = Depends(get_current_user_id)): + return await call(joint_binding_service.start(payload.device_id, user, payload.action)) + + +@router.get("/devices/{device_id}") +async def overview(device_id: str, user: int = Depends(get_current_user_id)): + return await call(joint_binding_service.overview(device_id, user)) + + +@router.get("/devices/{device_id}/portal") +async def portal(device_id: str, user: int = Depends(get_current_user_id)): + return await call(joint_binding_service.portal(device_id, user)) + + +@router.get("/{operation_id}") +async def get(operation_id: UUID, user: int = Depends(get_current_user_id)): + return await call(joint_binding_service.get(str(operation_id), user)) + + +@router.post("/{operation_id}/prepare") +async def prepare(operation_id: UUID, user: int = Depends(get_current_user_id)): + return await call(joint_binding_service.prepare(str(operation_id), user)) + + +@router.post("/{operation_id}/check") +async def check(operation_id: UUID, user: int = Depends(get_current_user_id)): + return await call(joint_binding_service.check(str(operation_id), user)) + + +@router.post("/{operation_id}/result") +async def report(operation_id: UUID, payload: ResultRequest, user: int = Depends(get_current_user_id)): + return await call(joint_binding_service.report(str(operation_id), user, **payload.model_dump())) + + +@router.post("/{operation_id}/finish-unbind") +async def finish_unbind(operation_id: UUID, user: int = Depends(get_current_user_id)): + return await call(joint_binding_service.finish_unbind(str(operation_id), user)) diff --git a/talkingq-url/banban/service/joint_binding.py b/talkingq-url/banban/service/joint_binding.py new file mode 100644 index 0000000..64cc398 --- /dev/null +++ b/talkingq-url/banban/service/joint_binding.py @@ -0,0 +1,297 @@ +"""Durable orchestration for the two mini-programs. + +Only server_confirmed represents the 4G relation service's decision. +Client receipts never authorize local unbinding. +""" +from uuid import uuid4 + +from sqlalchemy import text + +from banban.dao.binding import BindingDAO +from banban.service.traffic_handoff import TrafficError, binding_relation, create_handoff, normalize_iccid +from config import settings +from services.database_service_base import DatabaseServiceBase + + +class JointBindingService(DatabaseServiceBase): + async def overview(self, device_id, user_id): + db = await self.get_session() + try: + binding = await self._binding(db, device_id, user_id) + if int(binding["status"]) != 1: + raise TrafficError("NOT_BOUND", "设备当前未绑定") + rows = (await db.execute(text("SELECT * FROM device_binding_operations WHERE binding_instance_id=:instance"), + {"instance": binding["binding_instance_id"]})).mappings().all() + operations = {r["action"]: self.public(r) for r in rows} + return {"device_id": device_id, "binding_instance_id": binding["binding_instance_id"], + "bind": operations.get("bind"), "unbind": operations.get("unbind")} + finally: + await db.close() + + async def portal(self, device_id, user_id): + overview = await self.overview(device_id, user_id) + if overview["bind"] and not overview["unbind"]: + await self.check(overview["bind"]["operation_id"], user_id) + db = await self.get_session() + try: + binding = await self._binding(db, device_id, user_id) + rows = (await db.execute(text("SELECT * FROM device_binding_operations WHERE binding_instance_id=:instance"), + {"instance": binding["binding_instance_id"]})).mappings().all() + operations = {r["action"]: r for r in rows} + linked = operations.get("bind") + if (int(binding["status"]) != 1 or "unbind" in operations or not linked + or linked["status"] != "completed" or linked["traffic_status"] != "server_confirmed"): + raise TrafficError("TRAFFIC_BIND_REQUIRED", "请先完成流量服务关联") + return {"target_app_id": settings.traffic_4g_target_app_id, "iccid": linked["iccid"], + "source_binding_id": f"banban:{binding['binding_instance_id']}"} + finally: + await db.close() + + async def record_iccid(self, device_id: str, value) -> None: + iccid = normalize_iccid(value) + if iccid is None: + raise TrafficError("ICCID_INVALID", "设备尚未上报有效流量卡信息") + db = await self.get_session() + try: + # Do not silently replace a SIM participating in a binding operation. + current = (await db.execute(text("SELECT iccid FROM device_settings WHERE device_id=:device_id FOR UPDATE"), + {"device_id": device_id})).mappings().first() + if current and current["iccid"] and current["iccid"] != iccid: + raise TrafficError("ICCID_CONFLICT", "设备流量卡信息发生变化，需要核实") + if current: + await db.execute(text("UPDATE device_settings SET iccid=:iccid WHERE device_id=:device_id"), + {"device_id": device_id, "iccid": iccid}) + else: + await db.execute(text("INSERT INTO device_settings (device_id, iccid) VALUES (:device_id, :iccid)"), + {"device_id": device_id, "iccid": iccid}) + await db.commit() + finally: + await db.close() + + async def _binding(self, db, device_id, user_id): + row = (await db.execute(text("SELECT * FROM device_bindings WHERE device_id=:device_id FOR UPDATE"), + {"device_id": device_id})).mappings().first() + if not row or int(row["owner_user_id"]) != user_id: + raise TrafficError("NOT_DEVICE_OWNER", "仅设备归属账号可操作", 403) + return row + + def _check_current(self, binding, operation): + if (binding["binding_instance_id"] != operation["binding_instance_id"] + or (int(binding["status"]) != 1 and not (operation["action"] == "unbind" and operation["status"] == "completed"))): + raise TrafficError("STALE_OPERATION", "设备绑定关系已变化，请重新进入") + + async def _operation(self, db, operation_id, user_id): + # Consistent lock order: binding first, operation second. + row = (await db.execute(text("SELECT * FROM device_binding_operations WHERE operation_id=:id AND owner_user_id=:user"), + {"id": operation_id, "user": user_id})).mappings().first() + if not row: + raise TrafficError("OPERATION_NOT_FOUND", "未找到该操作", 404) + binding = await self._binding(db, row["device_id"], user_id) + row = (await db.execute(text("SELECT * FROM device_binding_operations WHERE operation_id=:id FOR UPDATE"), + {"id": operation_id})).mappings().one() + self._check_current(binding, row) + return dict(row) + + @staticmethod + def public(row): + return {key: row[key] for key in ("operation_id", "device_id", "action", "status", "traffic_status", "last_error")} + + async def start(self, device_id, user_id, action): + if action not in {"bind", "unbind"}: + raise TrafficError("INVALID_ACTION", "操作类型无效", 400) + db = await self.get_session() + try: + binding = await self._binding(db, device_id, user_id) + if int(binding["status"]) != 1: + raise TrafficError("NOT_BOUND", "设备当前未绑定") + instance = binding["binding_instance_id"] + other = (await db.execute(text("SELECT * FROM device_binding_operations WHERE binding_instance_id=:instance AND action='unbind'"), + {"instance": instance})).mappings().first() + if action == "bind" and other: + raise TrafficError("UNBIND_PENDING", "设备正在解绑，请先完成解绑") + row = (await db.execute(text("SELECT * FROM device_binding_operations WHERE binding_instance_id=:instance AND action=:action"), + {"instance": instance, "action": action})).mappings().first() + if row is None: + row = {"operation_id": str(uuid4()), "binding_instance_id": instance, + "device_id": device_id, "owner_user_id": user_id, "action": action, + "traffic_status": "pending", "status": "pending", "last_error": None} + await db.execute(text("""INSERT INTO device_binding_operations + (operation_id, binding_instance_id, device_id, owner_user_id, action) + VALUES (:operation_id, :binding_instance_id, :device_id, :owner_user_id, :action)"""), row) + await db.commit() + return self.public(row) + finally: + await db.close() + + async def get(self, operation_id, user_id): + db = await self.get_session() + try: + return self.public(await self._operation(db, operation_id, user_id)) + finally: + await db.close() + + async def check(self, operation_id, user_id, *, close=False): + db = await self.get_session() + try: + row = await self._operation(db, operation_id, user_id) + if row["action"] == "unbind" and row["status"] == "completed": + return self.public(row) + if close and row["action"] != "unbind": + raise TrafficError("INVALID_ACTION", "仅解绑操作可关闭流量关联") + finally: + await db.close() + relation = await binding_relation(f"banban:{row['binding_instance_id']}", + f"bb-{'close' if close else 'query'}-{operation_id}", close=close) + db = await self.get_session() + try: + # Revalidate after the network call; an old result cannot affect a rebind. + row = await self._operation(db, operation_id, user_id) + if row["status"] == "completed" and row["action"] == "unbind": + return self.public(row) + if row["action"] == "bind": + other = (await db.execute(text("SELECT operation_id FROM device_binding_operations WHERE binding_instance_id=:instance AND action='unbind'"), + {"instance": row["binding_instance_id"]})).first() + if other: + raise TrafficError("UNBIND_PENDING", "设备正在解绑，请先完成解绑") + confirmed = relation["status"] == "bound" and relation["relationState"] == "open" + if confirmed and not row["iccid"]: + setting = (await db.execute(text("SELECT iccid FROM device_settings WHERE device_id=:device_id"), + {"device_id": row["device_id"]})).mappings().first() + row["iccid"] = normalize_iccid(setting["iccid"] if setting else None) + if not row["iccid"]: + raise TrafficError("ICCID_NOT_AVAILABLE", "设备尚未上报流量卡信息，请设备联网后重试") + else: + confirmed = relation["canUnbind"] + error = None if confirmed else ( + "HANDOFF_RELATION_UNKNOWN" if relation["status"] == "unknown" else + "TRAFFIC_UNBIND_REQUIRED" if row["action"] == "unbind" and relation["status"] == "bound" else + "HANDOFF_RELATION_CLOSED" if row["action"] == "bind" and relation["relationState"] == "closing" else + "TRAFFIC_RESULT_UNCONFIRMED") + row.update(traffic_status="server_confirmed" if confirmed else "pending", + status="completed" if confirmed and row["action"] == "bind" else "pending", + last_error=error) + await db.execute(text("UPDATE device_binding_operations SET iccid=:iccid,traffic_status=:traffic_status,status=:status,last_error=:last_error WHERE operation_id=:operation_id"), row) + await db.commit() + return self.public(row) + finally: + await db.close() + + async def prepare(self, operation_id, user_id): + initial = await self.get(operation_id, user_id) + if initial["action"] == "unbind": + checked = await self.check(operation_id, user_id, close=True) + if checked["status"] == "completed" or checked["traffic_status"] == "server_confirmed": + return {"operation": checked} + if checked["last_error"] != "TRAFFIC_UNBIND_REQUIRED": + raise TrafficError(checked["last_error"] or "TRAFFIC_RESULT_UNCONFIRMED", + "流量关系尚未确认，伴伴绑定保留；请重试或联系服务人员") + elif initial["status"] == "completed": + checked = await self.check(operation_id, user_id) + if checked["status"] == "completed": + return {"operation": checked} + db = await self.get_session() + try: + row = await self._operation(db, operation_id, user_id) + if row["action"] == "bind": + unbinding = (await db.execute(text("SELECT operation_id FROM device_binding_operations WHERE binding_instance_id=:instance AND action='unbind'"), + {"instance": row["binding_instance_id"]})).first() + if unbinding: + raise TrafficError("UNBIND_PENDING", "设备正在解绑，请先完成解绑") + if row["status"] == "completed" or row["traffic_status"] == "server_confirmed": + return {"operation": self.public(row)} + setting = (await db.execute(text("SELECT iccid FROM device_settings WHERE device_id=:device_id"), + {"device_id": row["device_id"]})).mappings().first() + iccid = normalize_iccid(setting["iccid"] if setting else None) + if not iccid: + raise TrafficError("ICCID_NOT_AVAILABLE", "设备尚未上报流量卡信息，请设备联网后重试") + if row["iccid"] and row["iccid"] != iccid: + raise TrafficError("ICCID_CONFLICT", "设备流量卡信息发生变化，需要核实") + await db.execute(text("UPDATE device_binding_operations SET iccid=:iccid, handoff_started=1 WHERE operation_id=:id"), + {"id": operation_id, "iccid": iccid}) + await db.commit() + finally: + await db.close() + # Never hold a DB transaction across a network call or mini-program jump. + result = {"operation": self.public(row), "target_app_id": settings.traffic_4g_target_app_id, + "request_id": f"joint-{operation_id}", "handoff_token": None, + "source_binding_id": f"banban:{row['binding_instance_id']}"} + if row["action"] == "unbind": + # ICCID identifies the resource; only the 4G user's JWT authorizes unbind. + result["iccid"] = iccid + return result + request_id = f"bb-{operation_id}-{row['handoff_attempt']}" + try: + data = await create_handoff(iccid, result["source_binding_id"], request_id) + except TrafficError as exc: + if exc.code == "HANDOFF_TOKEN_USED": + checked = await self.check(operation_id, user_id) + if checked["status"] == "completed": + return {"operation": checked} + raise TrafficError(checked["last_error"] or "TRAFFIC_RESULT_UNCONFIRMED", + "原流量绑定结果尚未确认，请重试或联系服务人员") from None + if exc.code == "HANDOFF_TOKEN_EXPIRED": + db = await self.get_session() + try: + await self._operation(db, operation_id, user_id) + await db.execute(text("""UPDATE device_binding_operations + SET handoff_attempt=handoff_attempt+1 WHERE operation_id=:id AND handoff_attempt=:attempt"""), + {"id": operation_id, "attempt": row["handoff_attempt"]}) + await db.commit() + finally: + await db.close() + raise TrafficError(exc.code, "交接凭证已过期，请点击继续重新申请") from None + raise + result.update(handoff_token=data["handoffToken"], expires_in=data["expiresIn"]) + return result + + async def report(self, operation_id, user_id, action, request_id, iccid, code, success): + db = await self.get_session() + try: + row = await self._operation(db, operation_id, user_id) + if row["action"] != action or request_id != f"joint-{operation_id}": + raise TrafficError("RESULT_MISMATCH", "返回结果与当前操作不匹配") + if row["status"] == "completed" or row["traffic_status"] == "server_confirmed": + return self.public(row) + if success: + if code != "OK" or not row["iccid"] or normalize_iccid(iccid) != row["iccid"]: + raise TrafficError("RESULT_MISMATCH", "返回设备与当前操作不匹配") + row.update(traffic_status="client_confirmed", status="pending", last_error=None) + else: + row.update(status="partial_failed", last_error=code) + await db.execute(text("""UPDATE device_binding_operations SET traffic_status=:traffic_status, + status=:status, last_error=:last_error WHERE operation_id=:operation_id"""), row) + await db.commit() + finally: + await db.close() + # Even a success receipt must be checked against the original server relation. + return await self.check(operation_id, user_id) if success else self.public(row) + + async def finish_unbind(self, operation_id, user_id): + initial = await self.get(operation_id, user_id) + if initial["action"] != "unbind": + raise TrafficError("INVALID_ACTION", "仅解绑操作可完成解绑") + if initial["status"] == "completed": + return initial + checked = await self.check(operation_id, user_id, close=True) + if checked["traffic_status"] != "server_confirmed": + raise TrafficError(checked["last_error"] or "TRAFFIC_UNBIND_REQUIRED", + "流量侧尚未确认可解绑，伴伴绑定保留") + db = await self.get_session() + try: + row = await self._operation(db, operation_id, user_id) + if row["action"] != "unbind" or row["traffic_status"] != "server_confirmed": + raise TrafficError("TRAFFIC_UNBIND_REQUIRED", "请先完成流量管理侧解绑") + if row["status"] == "completed": + return self.public(row) + if not await BindingDAO(db).unbind(row["device_id"], user_id, commit=False): + raise TrafficError("NOT_BOUND", "设备绑定关系已变化") + await db.execute(text("UPDATE device_binding_operations SET status='completed', last_error=NULL WHERE operation_id=:id"), + {"id": operation_id}) + await db.commit() + row.update(status="completed", last_error=None) + return self.public(row) + finally: + await db.close() + + +joint_binding_service = JointBindingService("joint_binding") diff --git a/talkingq-url/banban/service/traffic_handoff.py b/talkingq-url/banban/service/traffic_handoff.py new file mode 100644 index 0000000..52285cf --- /dev/null +++ b/talkingq-url/banban/service/traffic_handoff.py @@ -0,0 +1,89 @@ +"""4G contract adapter. No user JWT or remote response body is logged.""" +import re +from datetime import datetime +from urllib.parse import quote + +import httpx + +from config import settings + + +class TrafficError(ValueError): + def __init__(self, code: str, message: str, status_code: int = 409): + super().__init__(message) + self.code = code + self.status_code = status_code + + +def normalize_iccid(value) -> str | None: + if not isinstance(value, str): + return None + value = value.strip().upper() + return value if re.fullmatch(r"[A-Z0-9]{18,22}", value) else None + + +async def _request(method: str, path: str, request_id: str, payload=None) -> dict: + base = settings.traffic_4g_api_base_url.rstrip("/") + if not base.startswith("https://") or not settings.banban_handoff_service_token: + raise TrafficError("CONFIGURATION_ERROR", "流量服务尚未配置，请稍后继续绑定", 503) + try: + async with httpx.AsyncClient(timeout=settings.traffic_4g_timeout_seconds) as client: + response = await client.request( + method, base + path, + headers={"X-Service-Token": settings.banban_handoff_service_token, + "X-Request-Id": request_id, "Cache-Control": "no-cache"}, + **({"json": payload} if payload is not None else {}), + ) + body = response.json() + except (httpx.HTTPError, ValueError): + raise TrafficError("TRAFFIC_UNAVAILABLE", "流量服务暂时无法确认结果，请重试", 503) from None + if not isinstance(body, dict): + raise TrafficError("INVALID_RESPONSE", "流量服务返回异常，请重试", 502) + if response.status_code != 200 or body.get("code") != "OK": + code = str(body.get("code") or "TRAFFIC_UNAVAILABLE") + allowed = {"HANDOFF_TOKEN_USED", "HANDOFF_TOKEN_EXPIRED", "DEVICE_DISABLED", + "IDEMPOTENCY_CONFLICT", "HANDOFF_DISABLED", "RATE_LIMITED", + "HANDOFF_SERVICE_UNAUTHORIZED", "INVALID_REQUEST", "CONFIGURATION_ERROR", + "HANDOFF_RELATION_CLOSED", "HANDOFF_RELATION_UNKNOWN"} + raise TrafficError(code if code in allowed else "TRAFFIC_UNAVAILABLE", + "流量绑定尚未完成，请重试或联系服务人员", 503 if response.status_code >= 500 else 409) + data = body.get("data") + if not isinstance(data, dict): + raise TrafficError("INVALID_RESPONSE", "流量服务返回异常，请重试", 502) + return data + + +async def create_handoff(iccid: str, source_binding_id: str, request_id: str) -> dict: + data = await _request("POST", "/internal/v1/handoff-tokens", request_id, + {"iccid": iccid, "sourceBindingId": source_binding_id, "requestId": request_id}) + if (not isinstance(data, dict) or data.get("targetAppId") != settings.traffic_4g_target_app_id + or not isinstance(data.get("handoffToken"), str) or not 1 <= len(data["handoffToken"]) <= 256 + or type(data.get("expiresIn")) is not int or data["expiresIn"] <= 0): + raise TrafficError("INVALID_RESPONSE", "流量服务返回异常，请重试", 502) + return data + + +async def binding_relation(source_binding_id: str, request_id: str, *, close=False) -> dict: + path = "/internal/v1/binding-relations/" + quote(source_binding_id, safe="") + data = await _request("POST" if close else "GET", path + ("/close" if close else ""), + request_id, {"requestId": request_id} if close else None) + valid = (data.get("sourceBindingId") == source_binding_id + and isinstance(data.get("status"), str) and data["status"] in {"bound", "unbound", "never_bound", "unknown"} + and isinstance(data.get("relationState"), str) and data["relationState"] in {"open", "closing", "unknown"} + and all(type(data.get(key)) is bool for key in ("pendingHandoff", "acceptingHandoff", "canUnbind"))) + expected = (data.get("relationState") == "closing" and data.get("pendingHandoff") is False + and data.get("acceptingHandoff") is False and data.get("status") in ("unbound", "never_bound")) + valid = valid and data.get("canUnbind") == expected + if data.get("relationState") == "closing": + valid = valid and not data.get("pendingHandoff") and not data.get("acceptingHandoff") + if data.get("relationState") == "unknown": + valid = valid and data.get("status") == "unknown" and not data.get("acceptingHandoff") + try: + checked = data.get("checkedAt") + valid = valid and isinstance(checked, str) and checked.endswith("Z") + datetime.fromisoformat(checked.replace("Z", "+00:00")) + except (AttributeError, TypeError, ValueError): + valid = False + if not valid: + raise TrafficError("INVALID_RESPONSE", "流量关系返回不完整或不一致，请重试", 502) + return data diff --git a/talkingq-url/config.py b/talkingq-url/config.py index a6f9e28..acdf735 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -24,6 +24,11 @@ class Settings(BaseSettings): server_host: str = Field(default="0.0.0.0", validation_alias="SERVER_HOST") server_port: int = Field(default=8080, validation_alias="SERVER_PORT") + traffic_4g_api_base_url: str = Field(default="https://role.api.talkingq.com/traffic-4g", validation_alias="TRAFFIC_4G_API_BASE_URL") + banban_handoff_service_token: str = Field(default="", validation_alias="BANBAN_HANDOFF_SERVICE_TOKEN") + traffic_4g_target_app_id: str = Field(default="wx56e1cfc52aab630f", validation_alias="TRAFFIC_4G_TARGET_APP_ID") + traffic_4g_timeout_seconds: float = Field(default=6.0, gt=0, le=15, validation_alias="TRAFFIC_4G_TIMEOUT_SECONDS") + asr_provider: str = Field(default="Aliyun", validation_alias="ASR_PROVIDER") llm_provider: str = Field(default="Volcano", validation_alias="LLM_PROVIDER") tts_provider: str = Field(default="MiniMax", validation_alias="TTS_PROVIDER") diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 1c6a904..ae73214 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -399,8 +399,24 @@ class TalkingQMQTTService: params = payload status = params.get("status") logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片返回状态: {status}") - data = params.get("data", {}) + if status != "success": + return + data = params.get("data") + if not isinstance(data, dict) or not isinstance(data.get("uuid"), str) or not data["uuid"].strip(): + logger.warning(device_id, "", "[NFC bind] invalid card response") + return nfc_uuid = data.get("uuid") + from banban.service.traffic_handoff import normalize_iccid + logger.info(device_id, "", f"[NFC bind] 006 ICCID present={'iccid' in data}, valid={normalize_iccid(data.get('iccid')) is not None}") + if data.get("iccid") is not None: + from banban.service.joint_binding import joint_binding_service + try: + await asyncio.wait_for(joint_binding_service.record_iccid(device_id, data["iccid"]), timeout=2) + logger.info(device_id, "", "[NFC bind] 006 ICCID persisted") + except Exception: + # A SIM persistence failure must not lose the NFC receipt. The + # combined operation remains pending until ICCID is available. + logger.warning(device_id, "", "[NFC bind] ICCID persistence failed; traffic binding pending") # 卡片与设备绑定 # await card_service.activate_card(device_id=device_id, card_uuid=nfc_uuid) result = await binding_service.finalize_nfc_bind(device_id=device_id, card_uuid=nfc_uuid) diff --git a/talkingq-url/mysql/init/02-init.sql b/talkingq-url/mysql/init/02-init.sql index 62c0c11..3c76229 100644 --- a/talkingq-url/mysql/init/02-init.sql +++ b/talkingq-url/mysql/init/02-init.sql @@ -312,6 +312,7 @@ CREATE TABLE IF NOT EXISTS `device_bindings` ( `id` BIGINT NOT NULL AUTO_INCREMENT, `device_id` VARCHAR(64) NOT NULL, `owner_user_id` BIGINT NOT NULL, + `binding_instance_id` CHAR(36) NULL, `child_id` BIGINT NULL, `status` TINYINT NOT NULL DEFAULT 1, `bound_at` DATETIME NOT NULL, @@ -320,6 +321,7 @@ CREATE TABLE IF NOT EXISTS `device_bindings` ( `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), CONSTRAINT `uq_device_binding_device` UNIQUE (`device_id`), + UNIQUE KEY `uk_binding_instance` (`binding_instance_id`), CONSTRAINT `uq_device_binding_child` UNIQUE (`child_id`), KEY `idx_device_bindings_owner_user_id` (`owner_user_id`), CONSTRAINT `fk_device_bindings_device` @@ -498,6 +500,7 @@ CREATE TABLE IF NOT EXISTS `device_settings` ( `signal` TINYINT UNSIGNED NULL, `version` VARCHAR(64) NULL, `imei` VARCHAR(64) NULL, + `iccid` VARCHAR(22) NULL, `device_type` TINYINT NULL, `ota_channel` VARCHAR(32) NULL, `disable_weekdays` VARCHAR(32) NULL, @@ -505,6 +508,7 @@ CREATE TABLE IF NOT EXISTS `device_settings` ( `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`setting_id`), UNIQUE KEY `uq_device_settings_device_id` (`device_id`), + UNIQUE KEY `uk_device_settings_iccid` (`iccid`), CONSTRAINT `fk_device_settings_device` FOREIGN KEY (`device_id`) REFERENCES `device_auth` (`device_id`) diff --git a/talkingq-url/mysql/init/03-joint-binding.sql b/talkingq-url/mysql/init/03-joint-binding.sql new file mode 100644 index 0000000..563dedb --- /dev/null +++ b/talkingq-url/mysql/init/03-joint-binding.sql @@ -0,0 +1,17 @@ +CREATE TABLE IF NOT EXISTS device_binding_operations ( + operation_id CHAR(36) NOT NULL PRIMARY KEY, + binding_instance_id CHAR(36) NOT NULL, + device_id VARCHAR(64) NOT NULL, + owner_user_id BIGINT NOT NULL, + action VARCHAR(8) NOT NULL, + iccid VARCHAR(22) NULL, + traffic_status VARCHAR(24) NOT NULL DEFAULT 'pending', + status VARCHAR(24) NOT NULL DEFAULT 'pending', + handoff_attempt INT NOT NULL DEFAULT 0, + handoff_started TINYINT NOT NULL DEFAULT 0, + last_error VARCHAR(64) NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uk_binding_operation (binding_instance_id, action), + KEY idx_binding_operation_owner (owner_user_id, device_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/talkingq-url/mysql/migrations/20260923_joint_binding.sql b/talkingq-url/mysql/migrations/20260923_joint_binding.sql new file mode 100644 index 0000000..d0e884e --- /dev/null +++ b/talkingq-url/mysql/migrations/20260923_joint_binding.sql @@ -0,0 +1,24 @@ +-- Run once on an existing database before deploying the application. +ALTER TABLE device_settings ADD COLUMN iccid VARCHAR(22) NULL, + ADD UNIQUE KEY uk_device_settings_iccid (iccid); +ALTER TABLE device_bindings ADD COLUMN binding_instance_id CHAR(36) NULL; +UPDATE device_bindings SET binding_instance_id = UUID() WHERE binding_instance_id IS NULL; +ALTER TABLE device_bindings ADD UNIQUE KEY uk_binding_instance (binding_instance_id); + +CREATE TABLE IF NOT EXISTS device_binding_operations ( + operation_id CHAR(36) NOT NULL PRIMARY KEY, + binding_instance_id CHAR(36) NOT NULL, + device_id VARCHAR(64) NOT NULL, + owner_user_id BIGINT NOT NULL, + action VARCHAR(8) NOT NULL, + iccid VARCHAR(22) NULL, + traffic_status VARCHAR(24) NOT NULL DEFAULT 'pending', + status VARCHAR(24) NOT NULL DEFAULT 'pending', + handoff_attempt INT NOT NULL DEFAULT 0, + handoff_started TINYINT NOT NULL DEFAULT 0, + last_error VARCHAR(64) NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uk_binding_operation (binding_instance_id, action), + KEY idx_binding_operation_owner (owner_user_id, device_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/talkingq-url/tests/test_joint_binding.py b/talkingq-url/tests/test_joint_binding.py new file mode 100644 index 0000000..cc302d3 --- /dev/null +++ b/talkingq-url/tests/test_joint_binding.py @@ -0,0 +1,347 @@ +"""Local SQL/state-machine checks; SQLite does not validate MySQL row locks.""" +import asyncio +import sqlite3 +from uuid import uuid4 + +import httpx +import pytest + +from banban.dao.binding import BindingDAO +from banban.service.joint_binding import JointBindingService +from banban.service.traffic_handoff import TrafficError, normalize_iccid +from banban.service import traffic_handoff + +DEVICE = "TalkingQ_test_joint" +ICCID = "898604E91123D1334901" + + +class Result: + def __init__(self, cursor): + self.rows = [dict(row) for row in cursor.fetchall()] if cursor.description else [] + def mappings(self): return self + def first(self): return self.rows[0] if self.rows else None + def one(self): + assert len(self.rows) == 1 + return self.rows[0] + def all(self): return self.rows + + +class Session: + def __init__(self, connection): self.connection = connection + async def execute(self, statement, params=None): + return Result(self.connection.execute(str(statement).replace(" FOR UPDATE", ""), params or {})) + async def commit(self): self.connection.commit() + async def rollback(self): self.connection.rollback() + async def close(self): self.connection.rollback() + + +@pytest.fixture +def state(monkeypatch): + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + conn.executescript(""" + CREATE TABLE device_bindings (id INTEGER PRIMARY KEY, device_id TEXT UNIQUE, owner_user_id INTEGER, + child_id INTEGER, status INTEGER, binding_instance_id TEXT, unbound_at TEXT, updated_at TEXT); + CREATE TABLE device_settings (device_id TEXT PRIMARY KEY, iccid TEXT UNIQUE); + CREATE TABLE device_binding_operations (operation_id TEXT PRIMARY KEY, binding_instance_id TEXT, + device_id TEXT, owner_user_id INTEGER, action TEXT, iccid TEXT, + traffic_status TEXT DEFAULT 'pending', status TEXT DEFAULT 'pending', + handoff_attempt INTEGER DEFAULT 0, handoff_started INTEGER DEFAULT 0, last_error TEXT, + UNIQUE(binding_instance_id, action)); + CREATE TABLE device_family_members (device_id TEXT, status INTEGER, removed_at TEXT, updated_at TEXT); + CREATE TABLE device_bind_history (device_id TEXT, child_id INTEGER, bound_by_user_id INTEGER, + unbound_by_user_id INTEGER, bind_source INTEGER, bound_at TEXT, unbound_at TEXT, unbind_reason TEXT); + """) + conn.execute("INSERT INTO device_bindings VALUES (1, ?, 7, NULL, 1, ?, NULL, NULL)", (DEVICE, str(uuid4()))) + conn.execute("INSERT INTO device_settings VALUES (?, ?)", (DEVICE, ICCID)) + conn.execute("INSERT INTO device_family_members VALUES (?, 1, NULL, NULL)", (DEVICE,)) + conn.commit() + service = JointBindingService() + service.test_relation = {"status": "bound", "relationState": "open"} + service.test_relation_calls = [] + async def relation(source, request_id, *, close=False): + service.test_relation_calls.append((source, request_id, close)) + if close: service.test_relation["relationState"] = "closing" + return {**service.test_relation, "sourceBindingId": source, "pendingHandoff": False, + "acceptingHandoff": service.test_relation["relationState"] == "open", + "canUnbind": service.test_relation["relationState"] == "closing" and service.test_relation["status"] in {"unbound", "never_bound"}} + monkeypatch.setattr("banban.service.joint_binding.binding_relation", relation) + async def session(): return Session(conn) + monkeypatch.setattr(service, "get_session", session) + yield service, conn + conn.close() + + +async def confirm(service, op, iccid=ICCID, success=True, code="OK"): + return await service.report(op["operation_id"], 7, op["action"], f"joint-{op['operation_id']}", iccid, code, success) + + +@pytest.mark.asyncio +async def test_bind_failure_preserves_banban_and_request_identity(state, monkeypatch): + service, conn = state + calls = [] + async def unavailable(*args): + calls.append(args) + raise TrafficError("TRAFFIC_UNAVAILABLE", "timeout", 503) + monkeypatch.setattr("banban.service.joint_binding.create_handoff", unavailable) + op = await service.start(DEVICE, 7, "bind") + for _ in range(2): + assert (await service.start(DEVICE, 7, "bind"))["operation_id"] == op["operation_id"] + with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7) + assert calls[0] == calls[1] + assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1 + assert (await service.overview(DEVICE, 7))["bind"]["status"] == "pending" + + +@pytest.mark.asyncio +async def test_missing_iccid_never_removes_binding_or_completes_unbind(state): + service, conn = state + conn.execute("UPDATE device_settings SET iccid=NULL") + conn.commit() + op = await service.start(DEVICE, 7, "unbind") + with pytest.raises(TrafficError, match="流量卡信息"): await service.prepare(op["operation_id"], 7) + with pytest.raises(TrafficError): await service.finish_unbind(op["operation_id"], 7) + assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1 + + +@pytest.mark.asyncio +async def test_successful_unbind_is_atomic_and_idempotent(state): + service, conn = state + op = await service.start(DEVICE, 7, "unbind") + await service.prepare(op["operation_id"], 7) + with pytest.raises(TrafficError): await service.finish_unbind(op["operation_id"], 7) + result = await confirm(service, op, success=False, code="NETWORK_ERROR") + assert result["status"] == "partial_failed" + with pytest.raises(TrafficError): await service.finish_unbind(op["operation_id"], 7) + service.test_relation["status"] = "unbound" + await confirm(service, op) + for _ in range(2): + result = await service.finish_unbind(op["operation_id"], 7) + assert result["status"] == "completed" + assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 0 + assert conn.execute("SELECT status FROM device_family_members").fetchone()[0] == 0 + + +@pytest.mark.asyncio +async def test_local_unbind_failure_rolls_back_and_keeps_4g_receipt(state, monkeypatch): + service, conn = state + op = await service.start(DEVICE, 7, "unbind") + await service.prepare(op["operation_id"], 7) + service.test_relation["status"] = "unbound" + await confirm(service, op) + original = BindingDAO.unbind + async def fail_after_update(self, *args, **kwargs): + await original(self, *args, **kwargs) + raise RuntimeError("local transaction failure") + monkeypatch.setattr(BindingDAO, "unbind", fail_after_update) + with pytest.raises(RuntimeError): await service.finish_unbind(op["operation_id"], 7) + assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1 + assert (await service.get(op["operation_id"], 7))["traffic_status"] == "server_confirmed" + monkeypatch.setattr(BindingDAO, "unbind", original) + assert (await service.finish_unbind(op["operation_id"], 7))["status"] == "completed" + + +@pytest.mark.asyncio +async def test_owner_device_action_and_lifecycle_guards(state): + service, conn = state + with pytest.raises(TrafficError): await service.start(DEVICE, 8, "bind") + op = await service.start(DEVICE, 7, "unbind") + await service.prepare(op["operation_id"], 7) + with pytest.raises(TrafficError): await service.get(op["operation_id"], 8) + with pytest.raises(TrafficError): await confirm(service, op, iccid="898604E91123D1334999") + with pytest.raises(TrafficError): await service.report(op["operation_id"], 7, "bind", f"joint-{op['operation_id']}", ICCID, "OK", True) + with pytest.raises(TrafficError): await service.report(op["operation_id"], 7, "unbind", "other-request", ICCID, "OK", True) + conn.execute("UPDATE device_bindings SET binding_instance_id=?", (str(uuid4()),)) + conn.commit() + with pytest.raises(TrafficError): await confirm(service, op) + + +@pytest.mark.asyncio +async def test_consumed_token_recovers_only_the_original_server_relation(state, monkeypatch): + service, _ = state + async def consumed(*args): raise TrafficError("HANDOFF_TOKEN_USED", "used") + monkeypatch.setattr("banban.service.joint_binding.create_handoff", consumed) + op = await service.start(DEVICE, 7, "bind") + result = await service.prepare(op["operation_id"], 7) + assert result["operation"]["status"] == "completed" + assert result["operation"]["traffic_status"] == "server_confirmed" + assert "recovery_only" not in result + assert (await service.portal(DEVICE, 7))["iccid"] == ICCID + await service.start(DEVICE, 7, "unbind") + with pytest.raises(TrafficError): await service.portal(DEVICE, 7) + with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["unbound", "never_bound"]) +async def test_already_released_relation_finishes_without_client_receipt_or_iccid(state, status): + service, conn = state + conn.execute("UPDATE device_settings SET iccid=NULL") + conn.commit() + service.test_relation["status"] = status + op = await service.start(DEVICE, 7, "unbind") + checked = await service.check(op["operation_id"], 7) + assert checked["traffic_status"] == "pending" # A plain GET on an open relation cannot authorize unbinding. + prepared = await service.prepare(op["operation_id"], 7) + assert prepared["operation"]["traffic_status"] == "server_confirmed" + assert "target_app_id" not in prepared # No redundant jump to the 4G mini-program. + assert (await service.finish_unbind(op["operation_id"], 7))["status"] == "completed" + assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 0 + close_calls = [call for call in service.test_relation_calls if call[2]] + assert len(set(close_calls)) == 1 + + +@pytest.mark.asyncio +async def test_unknown_relation_and_legacy_success_receipt_never_unbind(state): + service, conn = state + service.test_relation["status"] = "unknown" + op = await service.start(DEVICE, 7, "unbind") + conn.execute("UPDATE device_binding_operations SET traffic_status='client_confirmed'") + conn.commit() + with pytest.raises(TrafficError) as failure: await service.finish_unbind(op["operation_id"], 7) + assert failure.value.code == "HANDOFF_RELATION_UNKNOWN" + assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1 + with pytest.raises(TrafficError) as failure: await service.start(DEVICE, 7, "bind") + assert failure.value.code == "UNBIND_PENDING" + + +@pytest.mark.asyncio +async def test_close_network_failure_is_bounded_and_preserves_local_unbinding_marker(state, monkeypatch): + service, conn = state + calls = [] + async def unavailable(source, request_id, *, close=False): + calls.append((source, request_id, close)) + raise TrafficError("TRAFFIC_UNAVAILABLE", "timeout", 503) + monkeypatch.setattr("banban.service.joint_binding.binding_relation", unavailable) + op = await service.start(DEVICE, 7, "unbind") + for _ in range(2): + with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7) + assert calls[0] == calls[1] + assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1 + assert (await service.overview(DEVICE, 7))["unbind"]["operation_id"] == op["operation_id"] + + +@pytest.mark.asyncio +async def test_late_close_response_cannot_unbind_a_new_binding_instance(state, monkeypatch): + service, conn = state + async def changed(source, request_id, *, close=False): + conn.execute("UPDATE device_bindings SET binding_instance_id=?", (str(uuid4()),)) + conn.commit() + return {"status": "unbound", "canUnbind": True, "relationState": "closing"} + monkeypatch.setattr("banban.service.joint_binding.binding_relation", changed) + op = await service.start(DEVICE, 7, "unbind") + with pytest.raises(TrafficError) as failure: await service.finish_unbind(op["operation_id"], 7) + assert failure.value.code == "STALE_OPERATION" + assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1 + + +@pytest.mark.asyncio +async def test_consumed_token_unknown_or_released_relation_is_not_success(state, monkeypatch): + service, _ = state + async def consumed(*args): raise TrafficError("HANDOFF_TOKEN_USED", "used") + monkeypatch.setattr("banban.service.joint_binding.create_handoff", consumed) + op = await service.start(DEVICE, 7, "bind") + for status in ["unknown", "unbound", "never_bound"]: + service.test_relation["status"] = status + with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7) + assert (await service.get(op["operation_id"], 7))["status"] != "completed" + + +@pytest.mark.asyncio +async def test_successful_client_receipt_cannot_override_active_original_binding(state): + service, conn = state + op = await service.start(DEVICE, 7, "unbind") + await service.prepare(op["operation_id"], 7) + result = await confirm(service, op) + assert result["traffic_status"] == "pending" + with pytest.raises(TrafficError): await service.finish_unbind(op["operation_id"], 7) + assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1 + + +@pytest.mark.asyncio +async def test_expired_token_rotates_only_application_key(state, monkeypatch): + service, _ = state + calls = [] + async def expired(*args): + calls.append(args) + raise TrafficError("HANDOFF_TOKEN_EXPIRED", "expired") + monkeypatch.setattr("banban.service.joint_binding.create_handoff", expired) + op = await service.start(DEVICE, 7, "bind") + for _ in range(2): + with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7) + assert calls[0][:2] == calls[1][:2] + assert calls[0][2] != calls[1][2] + assert (await service.start(DEVICE, 7, "bind"))["operation_id"] == op["operation_id"] + + +def test_binding_generation_changes_only_for_new_ownership(): + row = {"status": 1, "owner_user_id": 7, "binding_instance_id": str(uuid4())} + assert BindingDAO._binding_instance(row, 7) == row["binding_instance_id"] + assert BindingDAO._binding_instance(row, 8) != row["binding_instance_id"] + row["status"] = 0 + assert BindingDAO._binding_instance(row, 7) != row["binding_instance_id"] + + +@pytest.mark.asyncio +async def test_iccid_normalization_and_conflict(state): + service, conn = state + assert normalize_iccid(ICCID.lower()) == ICCID + for invalid in [None, 1234, "123456789012345", "x" * 23, "#" * 20]: assert normalize_iccid(invalid) is None + await service.record_iccid(DEVICE, ICCID.lower()) + with pytest.raises(TrafficError): await service.record_iccid(DEVICE, "898604E91123D1334999") + assert conn.execute("SELECT iccid FROM device_settings").fetchone()[0] == ICCID + + +@pytest.mark.asyncio +async def test_mqtt_006_failure_and_iccid_failure(monkeypatch): + from handlers.mqtt_handler import TalkingQMQTTService + calls = [] + logs = [] + monkeypatch.setattr("handlers.mqtt_handler.logger.info", lambda *args: logs.append(args[-1])) + async def broken_iccid(*args): raise RuntimeError("database unavailable") + async def finalize(**kwargs): + calls.append(kwargs) + return None + monkeypatch.setattr("banban.service.joint_binding.joint_binding_service.record_iccid", broken_iccid) + monkeypatch.setattr("handlers.mqtt_handler.binding_service.finalize_nfc_bind", finalize) + mqtt = TalkingQMQTTService({"broker": "localhost"}) + await mqtt._handle_bind_response(DEVICE, {"status": "failed", "data": {"uuid": "CARD"}}) + await mqtt._handle_bind_response(DEVICE, {"status": "success", "data": None}) + assert not calls + await mqtt._handle_bind_response(DEVICE, {"status": "success", "data": {"uuid": "CARD", "iccid": ICCID}}) + await mqtt._handle_bind_response(DEVICE, {"status": "success", "data": {"uuid": "CARD"}}) + assert len(calls) == 2 + assert any("ICCID present=True, valid=True" in line for line in logs) + assert any("ICCID present=False, valid=False" in line for line in logs) + assert not any("ICCID persisted" in line for line in logs) + async def saved_iccid(*args): return None + monkeypatch.setattr("banban.service.joint_binding.joint_binding_service.record_iccid", saved_iccid) + await mqtt._handle_bind_response(DEVICE, {"status": "success", "data": {"uuid": "CARD", "iccid": ICCID}}) + assert any("ICCID persisted" in line for line in logs) + assert not any(ICCID in line for line in logs) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("body", [None, {"code": "OK", "data": {}}, + {"code": "OK", "data": {"targetAppId": "wrong", "handoffToken": "secret", "expiresIn": 300}}, + {"code": "OK", "data": {"targetAppId": "wx56e1cfc52aab630f", "handoffToken": "secret", "expiresIn": True}}]) +async def test_upstream_malformed_reply_is_not_success(monkeypatch, body): + real_client = httpx.AsyncClient + transport = httpx.MockTransport(lambda req: httpx.Response(200, json=body)) + monkeypatch.setattr(traffic_handoff.settings, "banban_handoff_service_token", "test-service-token") + monkeypatch.setattr(traffic_handoff.httpx, "AsyncClient", lambda **kwargs: real_client(transport=transport, **kwargs)) + with pytest.raises(TrafficError): await traffic_handoff.create_handoff(ICCID, "banban:test", "test-request") + + +@pytest.mark.asyncio +async def test_router_timeout_returns_retryable_error(monkeypatch): + from banban.routers.joint_binding import call + from fastapi import HTTPException + real_wait_for = asyncio.wait_for + async def short_wait(coro, timeout): + assert timeout == 8 + return await real_wait_for(coro, timeout=0.001) + monkeypatch.setattr(asyncio, "wait_for", short_wait) + with pytest.raises(HTTPException) as failure: await call(asyncio.sleep(10)) + assert failure.value.status_code == 503 + assert failure.value.detail["code"] == "OPERATION_TIMEOUT" diff --git a/talkingq-url/tests/test_traffic_relation.py b/talkingq-url/tests/test_traffic_relation.py new file mode 100644 index 0000000..cb7ebad --- /dev/null +++ b/talkingq-url/tests/test_traffic_relation.py @@ -0,0 +1,79 @@ +import json + +import httpx +import pytest + +from banban.service import traffic_handoff as adapter + +SOURCE = "banban:12345678-1234-1234-1234-123456789abc" +RELEASED = {"sourceBindingId": SOURCE, "status": "unbound", "relationState": "closing", + "pendingHandoff": False, "acceptingHandoff": False, "canUnbind": True, + "reason": "ORIGINAL_BINDING_RELEASED", "checkedAt": "2026-09-23T08:00:00Z"} + + +def transport(monkeypatch, callback): + client = httpx.AsyncClient + monkeypatch.setattr(adapter.settings, "banban_handoff_service_token", "test-only") + monkeypatch.setattr(adapter.httpx, "AsyncClient", lambda **kwargs: client(transport=httpx.MockTransport(callback), **kwargs)) + + +@pytest.mark.asyncio +async def test_query_and_close_encode_existing_ids_and_use_service_auth(monkeypatch): + requests = [] + def reply(request): + requests.append(request) + return httpx.Response(200, json={"code": "OK", "data": RELEASED}) + transport(monkeypatch, reply) + await adapter.binding_relation(SOURCE, "query-1") + for _ in range(2): await adapter.binding_relation(SOURCE, "close-1", close=True) + assert requests[0].method == "GET" and requests[0].content == b"" + assert b"banban%3A" in requests[0].url.raw_path + assert all(request.headers["X-Service-Token"] == "test-only" for request in requests) + assert all("Authorization" not in request.headers for request in requests) + assert requests[0].headers["Cache-Control"] == "no-cache" + assert requests[1].url.path.endswith("/close") and requests[1].method == "POST" + assert json.loads(requests[1].content) == {"requestId": "close-1"} + assert requests[1].content == requests[2].content + + +@pytest.mark.asyncio +@pytest.mark.parametrize("changes", [ + {"sourceBindingId": "another"}, {"canUnbind": "true"}, {"pendingHandoff": 0}, + {"pendingHandoff": True}, {"acceptingHandoff": True}, {"relationState": "open"}, + {"status": "bound"}, {"status": "unknown"}, {"status": {}}, + {"relationState": []}, {"checkedAt": None}, {"checkedAt": "badZ"}, + {"checkedAt": "2026-09-23T08:00:00"}, {"canUnbind": False}, +]) +async def test_inconsistent_or_malformed_relation_never_authorizes_unbind(monkeypatch, changes): + transport(monkeypatch, lambda request: httpx.Response(200, json={"code": "OK", "data": {**RELEASED, **changes}})) + with pytest.raises(adapter.TrafficError) as failure: await adapter.binding_relation(SOURCE, "query") + assert failure.value.code == "INVALID_RESPONSE" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status,code", [(401, "HANDOFF_SERVICE_UNAUTHORIZED"), (429, "RATE_LIMITED"), + (500, "INTERNAL_ERROR"), (404, "NOT_FOUND"), (200, "HANDOFF_RELATION_UNKNOWN")]) +async def test_http_or_business_errors_never_become_unbound(monkeypatch, status, code): + transport(monkeypatch, lambda request: httpx.Response(status, json={"code": code, "data": RELEASED})) + with pytest.raises(adapter.TrafficError): await adapter.binding_relation(SOURCE, "close", close=True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("state", [ + {**RELEASED, "status": "never_bound"}, + {**RELEASED, "status": "bound", "canUnbind": False}, + {**RELEASED, "status": "unknown", "canUnbind": False}, + {**RELEASED, "status": "unknown", "relationState": "unknown", "canUnbind": False}, + {**RELEASED, "status": "never_bound", "relationState": "open", "acceptingHandoff": True, "pendingHandoff": True, "canUnbind": False}, +]) +async def test_valid_non_success_business_states_are_preserved(monkeypatch, state): + transport(monkeypatch, lambda request: httpx.Response(200, json={"code": "OK", "data": state})) + assert await adapter.binding_relation(SOURCE, "query") == state + + +@pytest.mark.asyncio +async def test_relation_network_timeout_is_explicit(monkeypatch): + def timeout(request): raise httpx.ReadTimeout("test timeout") + transport(monkeypatch, timeout) + with pytest.raises(adapter.TrafficError) as failure: await adapter.binding_relation(SOURCE, "query") + assert failure.value.code == "TRAFFIC_UNAVAILABLE"