From 491b53184156867773fbab401036c5eb9965ddef Mon Sep 17 00:00:00 2001 From: DrSmoothl <1787882683@qq.com> Date: Wed, 21 Oct 2026 16:59:26 +0800 Subject: [PATCH] feat: support client private-message reporting Handle confirmed private messages sent from the client and preserve account isolation when sessions change. Surface unavailable messages without reporting incomplete content. Refs #528. --- packages/bridge/src/hook-session.ts | 12 + packages/bridge/src/qq-hook-client.ts | 42 +++ packages/bridge/tests/hook-session.test.ts | 36 +++ packages/bridge/tests/qq-hook-client.test.ts | 108 +++++++ packages/common/src/protocol-types.ts | 3 + packages/core/src/bridge/bridge.ts | 89 +++++- .../core/tests/bridge-client-message.test.ts | 207 +++++++++++++ packages/proto-defs/src/action.ts | 8 + .../src/msg-push/client-message-receipt.ts | 157 ++++++++++ packages/protocol/src/msg-push/index.ts | 1 + packages/protocol/src/packet-pipeline.ts | 12 +- .../msg-push/client-message-receipt.test.ts | 292 ++++++++++++++++++ 12 files changed, 963 insertions(+), 4 deletions(-) create mode 100644 packages/core/tests/bridge-client-message.test.ts create mode 100644 packages/protocol/src/msg-push/client-message-receipt.ts create mode 100644 packages/protocol/tests/msg-push/client-message-receipt.test.ts diff --git a/packages/bridge/src/hook-session.ts b/packages/bridge/src/hook-session.ts index 115a308c..4b975607 100644 --- a/packages/bridge/src/hook-session.ts +++ b/packages/bridge/src/hook-session.ts @@ -957,6 +957,17 @@ export class HookSession extends EventEmitter { return; } + if ( + packet.source === 'client-message' && + (!isRealUin(packet.uin) || packet.uin !== this._uin) + ) { + this.log.warn( + 'Client message receipt belongs to an inactive account: PID=%d', + this.pid, + ); + return; + } + const uin = packet.uin || this._uin; // Routing requires a real UIN, but even a frame carrying UIN=0 proves the // receive hook is alive. Attribute that evidence to the current epoch @@ -986,6 +997,7 @@ export class HookSession extends EventEmitter { seqId: packet.seq, retCode: packet.error, fromClient: false, + ...(packet.source ? { source: packet.source } : {}), body: Buffer.from(packet.body), }); } diff --git a/packages/bridge/src/qq-hook-client.ts b/packages/bridge/src/qq-hook-client.ts index 5723ba00..c763eba0 100644 --- a/packages/bridge/src/qq-hook-client.ts +++ b/packages/bridge/src/qq-hook-client.ts @@ -7,6 +7,7 @@ import { } from '@snowluma/common/logger'; import { renderParamsVerbose } from '@snowluma/common/log-summary'; import { isRealUin } from '@snowluma/common/uin'; +import { CLIENT_MESSAGE_RECEIPT_CMD } from '@snowluma/common/protocol-types'; import { readdirSync, promises as fs } from 'fs'; import net from 'net'; import path from 'path'; @@ -38,6 +39,8 @@ export enum PipeOp { recvPacket = 6, loginState = 7, loginIdentityHint = 17, + clientMessageReceipt = 21, + clientMessageState = 22, } const PipeFlagWantReply = 1 << 0; @@ -55,6 +58,7 @@ export interface QqHookPacket { cmd: string; uin: string; body: Buffer; + source?: 'client-message'; } export interface QqHookHello { @@ -1369,6 +1373,44 @@ export class QqHookClient extends EventEmitter { return; } + if (frame.op === PipeOp.clientMessageState) { + if (frame.status === 1) { + runtimeLog.debug('Client private-message reporting is available: PID=%d', this.pid); + } else { + runtimeLog.warn( + 'Client private-message reporting is unavailable: PID=%d status=%d', + this.pid, + frame.status, + ); + } + + return; + } + + if (frame.op === PipeOp.clientMessageReceipt) { + if ( + frame.cmd !== '' || + frame.requestId !== 0 || + frame.flags !== 0 || + frame.value0 <= 0n || + frame.value0 > BigInt(Number.MAX_SAFE_INTEGER) || + !isRealUin(frame.msg) + ) { + runtimeLog.warn('Invalid client message receipt: PID=%d', this.pid); + return; + } + + this.emit('packet', { + seq: Number(frame.value0), + error: frame.status, + cmd: CLIENT_MESSAGE_RECEIPT_CMD, + uin: frame.msg, + body: Buffer.from(frame.body), + source: 'client-message', + } satisfies QqHookPacket); + return; + } + if (frame.op !== PipeOp.recvPacket) { return; } diff --git a/packages/bridge/tests/hook-session.test.ts b/packages/bridge/tests/hook-session.test.ts index 5e9417e1..59680abb 100644 --- a/packages/bridge/tests/hook-session.test.ts +++ b/packages/bridge/tests/hook-session.test.ts @@ -929,6 +929,42 @@ describe('HookSession — pipe-down must not clobber a settled error', () => { }); describe('HookSession — packet forwarding', () => { + it('forwards client receipts only for the active account without changing direction', async () => { + const onPacket = vi.fn(); + const ctx = makeSession({ pipeLive: true, onPacket }); + await ctx.session.load(); + ctx.session.onPipeUp(); + await flush(); + ctx.currentClient().fireLogin('10001'); + const receipt: QqHookPacket = { + seq: 42, + error: 0, + cmd: 'ClientMessageReceipt', + uin: '10001', + source: 'client-message', + body: Buffer.from([1, 2, 3]), + }; + + for (const uin of ['', '0', '20002']) { + ctx.currentClient().firePacket({ ...receipt, uin }); + } + + expect(onPacket).not.toHaveBeenCalled(); + ctx.currentClient().firePacket(receipt); + expect(onPacket).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + pid: 1234, + uin: '10001', + source: 'client-message', + fromClient: false, + }), + ); + ctx.currentClient().fireLogin('20002'); + ctx.currentClient().firePacket(receipt); + expect(onPacket).toHaveBeenCalledOnce(); + await ctx.session.dispose(); + }); + // These tests cover the field-rename + filter logic that used to live // inside the deleted NtqqHandler.onHookPacket. Inlining it here means // we can exercise the transformation with a plain vi.fn() sink, no diff --git a/packages/bridge/tests/qq-hook-client.test.ts b/packages/bridge/tests/qq-hook-client.test.ts index 037ca16f..36e35384 100644 --- a/packages/bridge/tests/qq-hook-client.test.ts +++ b/packages/bridge/tests/qq-hook-client.test.ts @@ -12,6 +12,7 @@ import { type LogEntry, } from '@snowluma/common/logger'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import { CLIENT_MESSAGE_RECEIPT_CMD } from '@snowluma/common/protocol-types'; import { HEADER_SIZE, listLiveLinuxPipePids, @@ -147,6 +148,113 @@ describe('QqHookClient login wait lifecycle', () => { }); }); +describe('QqHookClient client message receipts', () => { + it('preserves receipt provenance and failure status separately from normal packets', async () => { + const sockets = mockControlPipe(12345, () => {}); + const client = new QqHookClient(12345); + const onPacket = vi.fn(); + + try { + await client.startRecv(onPacket); + const socket = sockets.mock.results[0]!.value; + const body = Buffer.from([1, 2, 3]); + socket.emit( + 'data', + pipeFrame({ op: PipeOp.clientMessageReceipt, value0: 42n, message: '10001', body }), + ); + socket.emit( + 'data', + pipeFrame({ + op: PipeOp.clientMessageReceipt, + value0: 43n, + message: '10001', + status: -1, + }), + ); + socket.emit( + 'data', + pipeFrame({ + op: PipeOp.recvPacket, + value0: 44n, + message: '10001', + cmd: CLIENT_MESSAGE_RECEIPT_CMD, + body, + }), + ); + + expect(onPacket.mock.calls.map(([packet]) => packet)).toEqual([ + { + seq: 42, + error: 0, + cmd: CLIENT_MESSAGE_RECEIPT_CMD, + uin: '10001', + body, + source: 'client-message', + }, + { + seq: 43, + error: -1, + cmd: CLIENT_MESSAGE_RECEIPT_CMD, + uin: '10001', + body: Buffer.alloc(0), + source: 'client-message', + }, + { seq: 44, error: 0, cmd: CLIENT_MESSAGE_RECEIPT_CMD, uin: '10001', body }, + ]); + } finally { + client.close(); + } + }); + + it('reports unsupported capability and invalid envelopes without breaking normal receive', async () => { + const entries: LogEntry[] = []; + setLogLevel('warn'); + const unsubscribe = subscribeLogs((entry) => entries.push(entry)); + const sockets = mockControlPipe(12345, () => {}); + const client = new QqHookClient(12345); + const onPacket = vi.fn(); + + try { + await client.startRecv(onPacket); + const socket = sockets.mock.results[0]!.value; + socket.emit('data', pipeFrame({ op: PipeOp.clientMessageState, status: -1 })); + + for (const patch of [ + { value0: 0n }, + { value0: BigInt(Number.MAX_SAFE_INTEGER) + 1n }, + { message: '' }, + { cmd: 'unexpected' }, + { requestId: 1 }, + ]) { + socket.emit( + 'data', + pipeFrame({ + op: PipeOp.clientMessageReceipt, + value0: 42n, + message: '10001', + ...patch, + }), + ); + } + + expect(onPacket).not.toHaveBeenCalled(); + expect( + entries.filter( + (entry) => entry.scope === 'QQHook.Runtime' && entry.level === 'warn', + ), + ).toHaveLength(6); + socket.emit( + 'data', + pipeFrame({ op: PipeOp.recvPacket, cmd: 'normal', message: '10001', value0: 44n }), + ); + expect(onPacket).toHaveBeenCalledOnce(); + } finally { + client.close(); + unsubscribe(); + } + }); +}); + describe('listLiveLinuxPipePids', () => { it('ignores stale-looking control socket names when the probe rejects them', async () => { const runtimeDir = await makeRuntimeDir(); diff --git a/packages/common/src/protocol-types.ts b/packages/common/src/protocol-types.ts index 590ecc60..2fa30902 100644 --- a/packages/common/src/protocol-types.ts +++ b/packages/common/src/protocol-types.ts @@ -1,3 +1,5 @@ +export const CLIENT_MESSAGE_RECEIPT_CMD = 'ClientMessageReceipt'; + export interface PacketInfo { pid: number; uin: string; @@ -6,6 +8,7 @@ export interface PacketInfo { seqId: number; retCode: number; fromClient: boolean; + source?: 'client-message'; body: Uint8Array; } export type PacketSink = (packet: PacketInfo) => void; diff --git a/packages/core/src/bridge/bridge.ts b/packages/core/src/bridge/bridge.ts index 0318e89b..5ea8073e 100644 --- a/packages/core/src/bridge/bridge.ts +++ b/packages/core/src/bridge/bridge.ts @@ -4,7 +4,13 @@ import { packetTimeoutMs } from '@snowluma/bridge/packet-timeout'; import type { PacketInfo } from '@snowluma/common/protocol-types'; import { BridgeEventBus } from '@snowluma/protocol/event-bus'; import { IdentityService } from '@snowluma/protocol/identity-service'; -import { MSG_PUSH_CMD, parseMsgPushOrThrow, SysMsgDedup } from '@snowluma/protocol/msg-push'; +import { + MSG_PUSH_CMD, + parseMsgPushOrThrow, + SysMsgDedup, + decodeClientMessageReceipt, + fetchClientMessage, +} from '@snowluma/protocol/msg-push'; import { KICK_NT_CMD, parseKickNT } from '@snowluma/protocol/kick-nt'; import { IncomingPacketPipeline, type CmdParser } from '@snowluma/protocol/packet-pipeline'; import type { OnlineDeviceInfo } from '@snowluma/protocol/events'; @@ -38,6 +44,10 @@ export class Bridge implements BridgeInterface { private onlineClients_: readonly Readonly[] | null = null; private disposePromise_: Promise | null = null; private readonly pendingPackets_ = new Set>(); + private readonly clientMessageTasks_ = new Map< + string, + { connection: ProcessConnection; work: Promise } + >(); private rosterWarmupStarted_ = false; private settleRosterWarmup!: (result: RosterWarmupResult) => void; private readonly rosterWarmup = new Promise((resolve) => { @@ -111,9 +121,13 @@ export class Bridge implements BridgeInterface { bindPid(pid: number, client: PacketSender): void { const previous = this.activeConnection; + const existing = this.connections.get(pid); this.connections.delete(pid); // A new login stays healthy until the watchdog confirms a failure. - this.connections.set(pid, { pid, client, receiveHealthy: true }); + const connection = + existing?.client === client ? existing : { pid, client, receiveHealthy: true }; + connection.receiveHealthy = true; + this.connections.set(pid, connection); if (previous?.pid !== pid || previous.client !== client) { runtimeLog.trace( @@ -190,12 +204,81 @@ export class Bridge implements BridgeInterface { return Promise.reject(new Error('Bridge is disposed')); } - const work = this.identity.initialization.then(() => this.pipeline.process(pkt)); + const connection = this.connections.get(pkt.pid); + const work = this.identity.initialization.then(() => + pkt.source === 'client-message' + ? this.processClientMessage(pkt, connection) + : this.pipeline.process(pkt), + ); this.pendingPackets_.add(work); void work.finally(() => this.pendingPackets_.delete(work)).catch(() => undefined); return work; } + private async processClientMessage( + pkt: PacketInfo, + connection: ProcessConnection | undefined, + ): Promise { + const requireConnection = (): ProcessConnection => { + if ( + this.disposePromise_ || + !connection || + this.connections.get(pkt.pid) !== connection + ) { + throw new Error('Client message receipt belongs to an inactive connection'); + } + + return connection; + }; + + const current = requireConnection(); + const receipt = decodeClientMessageReceipt(pkt, this.identity.uin); + const key = JSON.stringify([ + pkt.pid, + receipt.peerUid, + receipt.sequence, + receipt.clientSequence, + receipt.timestamp, + ]); + const pending = this.clientMessageTasks_.get(key); + + if (pending?.connection === current) { + return pending.work; + } + + if (this.clientMessageTasks_.size >= 256) { + throw new Error('Too many pending client message lookups'); + } + + runtimeLog.trace( + 'client_message_lookup_start uin=%j pid=%d seqId=%d', + this.identity.uin, + pkt.pid, + pkt.seqId, + ); + const work = (async () => { + const event = await fetchClientMessage(current.client, this.identity, receipt); + requireConnection(); + await this.pipeline.processEvents(pkt, [event]); + runtimeLog.trace( + 'client_message_lookup_complete uin=%j pid=%d seqId=%d', + this.identity.uin, + pkt.pid, + pkt.seqId, + ); + })(); + const task = { connection: current, work }; + this.clientMessageTasks_.set(key, task); + + try { + await work; + } finally { + if (this.clientMessageTasks_.get(key) === task) { + this.clientMessageTasks_.delete(key); + } + } + } + rememberUploadedFile(meta: UploadedFileMeta): void { if (!meta.fileId) { return; diff --git a/packages/core/tests/bridge-client-message.test.ts b/packages/core/tests/bridge-client-message.test.ts new file mode 100644 index 00000000..e184a3d0 --- /dev/null +++ b/packages/core/tests/bridge-client-message.test.ts @@ -0,0 +1,207 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { CLIENT_MESSAGE_RECEIPT_CMD, type PacketInfo } from '@snowluma/common/protocol-types'; +import type { PacketSender, SendPacketResult } from '@snowluma/common/packet-sender'; +import type { SendMessageResponse } from '@snowluma/proto-defs/action'; +import type { SsoGetC2cMsgResponse } from '@snowluma/proto-defs/get-c2c-msg'; +import { protobuf_encode } from '@snowluma/proton'; +import { IdentityService } from '@snowluma/protocol/identity-service'; +import { Bridge } from '../src/bridge/bridge'; + +const bridges: Bridge[] = []; +const timestamp = 1_700_000_000; + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((accept) => { + resolve = accept; + }); + return { promise, resolve }; +} + +afterEach(async () => { + await Promise.all(bridges.splice(0).map((bridge) => bridge.dispose())); + vi.restoreAllMocks(); +}); + +function packet(): PacketInfo { + const uid = Buffer.from('u_test_peer'); + const header = Buffer.alloc(8); + header[0] = 1; + header[1] = uid.length; + header.writeUInt32LE(9, 4); + return { + pid: 1234, + uin: '10001', + source: 'client-message', + serviceCmd: CLIENT_MESSAGE_RECEIPT_CMD, + seqId: 42, + retCode: 0, + fromClient: false, + body: Buffer.concat([ + header, + uid, + protobuf_encode({ privateSequence: 87, timestamp1: timestamp }), + ]), + }; +} + +function response(): SendPacketResult { + return { + success: true, + gotResponse: true, + errorCode: 0, + errorMessage: '', + responseData: Buffer.from( + protobuf_encode({ + friendUid: 'u_test_peer', + messages: [ + { + responseHead: { + fromUin: 10001, + fromUid: 'u_test_self', + toUin: 20002, + toUid: 'u_test_peer', + }, + contentHead: { + msgType: 166, + sequence: 9, + ntMsgSeq: 87, + timestamp, + msgId: 123456, + }, + body: { richText: { elems: [{ text: { str: 'hello' } }] } }, + }, + ], + }), + ), + }; +} + +async function setup() { + const bridge = new Bridge(IdentityService.memory('10001')); + bridges.push(bridge); + vi.spyOn(bridge.apis.contacts, 'fetchFriendList').mockResolvedValue([ + { uin: 10001, uid: 'u_test_self', nickname: 'self', remark: '' }, + ]); + vi.spyOn(bridge.apis.contacts, 'fetchGroupList').mockResolvedValue([]); + const sendPacket = vi.fn(async () => response()); + const sender = { sendPacket }; + bridge.bindPid(1234, sender); + await bridge.whenRosterWarmupSettled(); + const received = vi.fn(); + bridge.events.on('friend_message', received); + return { bridge, sendPacket, sender, received }; +} + +describe('Bridge client message recovery', () => { + it('dispatches confirmed content through the normal event pipeline', async () => { + const ctx = await setup(); + await ctx.bridge.onPacket(packet()); + expect(ctx.sendPacket).toHaveBeenCalledOnce(); + expect(ctx.received).toHaveBeenCalledOnce(); + expect(ctx.received).toHaveBeenCalledWith( + expect.objectContaining({ + kind: 'friend_message', + selfUin: 10001, + senderUin: 10001, + peerUin: 20002, + ntMsgSeq: 87, + clientSeq: 9, + elements: [{ type: 'text', text: 'hello' }], + }), + ); + }); + + it('does not treat an ordinary packet with the same command as a receipt', async () => { + const ctx = await setup(); + const ordinary = packet(); + delete ordinary.source; + await ctx.bridge.onPacket(ordinary); + expect(ctx.sendPacket).not.toHaveBeenCalled(); + expect(ctx.received).not.toHaveBeenCalled(); + }); + + it('shares an in-flight lookup across repeated receipts and ordinary PID observations', async () => { + const ctx = await setup(); + const result = deferred(); + const started = deferred(); + ctx.sendPacket.mockImplementation(() => { + started.resolve(); + return result.promise; + }); + const first = ctx.bridge.onPacket(packet()); + await started.promise; + ctx.bridge.bindPid(1234, ctx.sender); + const duplicate = ctx.bridge.onPacket(packet()); + await Promise.resolve(); + result.resolve(response()); + await Promise.all([first, duplicate]); + expect(ctx.sendPacket).toHaveBeenCalledOnce(); + expect(ctx.received).toHaveBeenCalledOnce(); + }); + + it.each(['detach', 'replace', 'reattach', 'dispose'] as const)( + 'rejects a lookup completed after %s', + async (action) => { + const ctx = await setup(); + const result = deferred(); + const started = deferred(); + ctx.sendPacket.mockImplementation(() => { + started.resolve(); + return result.promise; + }); + const work = ctx.bridge.onPacket(packet()); + const rejected = expect(work).rejects.toThrow('inactive connection'); + await started.promise; + let disposal: Promise | undefined; + + if (action === 'dispose') { + disposal = ctx.bridge.dispose(); + } else if (action === 'replace') { + ctx.bridge.bindPid(1234, { sendPacket: vi.fn() }); + } else { + ctx.bridge.detachPid(1234); + + if (action === 'reattach') { + ctx.bridge.bindPid(1234, ctx.sender); + } + } + + result.resolve(response()); + await rejected; + await disposal; + expect(ctx.received).not.toHaveBeenCalled(); + }, + ); + + it('queries the originating connection even when another process is active', async () => { + const ctx = await setup(); + const other = { sendPacket: vi.fn() }; + ctx.bridge.bindPid(4321, other); + await ctx.bridge.onPacket(packet()); + expect(ctx.sendPacket).toHaveBeenCalledOnce(); + expect(other.sendPacket).not.toHaveBeenCalled(); + }); + + it('rejects unavailable, detached and wrong-account receipts before querying', async () => { + const ctx = await setup(); + await expect(ctx.bridge.onPacket({ ...packet(), retCode: -1 })).rejects.toThrow( + 'unavailable', + ); + await expect(ctx.bridge.onPacket({ ...packet(), uin: '20002' })).rejects.toThrow('account'); + ctx.bridge.detachPid(1234); + await expect(ctx.bridge.onPacket(packet())).rejects.toThrow('inactive connection'); + expect(ctx.sendPacket).not.toHaveBeenCalled(); + expect(ctx.received).not.toHaveBeenCalled(); + }); + + it('surfaces lookup failure and allows a later receipt to retry', async () => { + const ctx = await setup(); + ctx.sendPacket.mockRejectedValueOnce(new Error('lookup failed')); + await expect(ctx.bridge.onPacket(packet())).rejects.toThrow('lookup failed'); + expect(ctx.received).not.toHaveBeenCalled(); + await ctx.bridge.onPacket(packet()); + expect(ctx.sendPacket).toHaveBeenCalledTimes(2); + expect(ctx.received).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/proto-defs/src/action.ts b/packages/proto-defs/src/action.ts index eff67f64..48729195 100644 --- a/packages/proto-defs/src/action.ts +++ b/packages/proto-defs/src/action.ts @@ -79,12 +79,20 @@ export interface SendMessageResponse { result?: pb<1, int_32>; errMsg?: pb<2, string>; timestamp1?: pb<3, uint_32>; + reviewResult?: pb<6, int_32>; + serverResult?: pb<7, SendMessageServerResult>; field10?: pb<10, uint_32>; groupSequence?: pb<11, uint_32>; timestamp2?: pb<12, uint_32>; privateSequence?: pb<14, uint_32>; } +export interface SendMessageServerResult { + type?: pb<1, int_32>; + code?: pb<2, int_32>; + message?: pb<3, string>; +} + // MentionExtra (for building @mention text element) export interface MentionExtraSend { type?: pb<3, int_32>; diff --git a/packages/protocol/src/msg-push/client-message-receipt.ts b/packages/protocol/src/msg-push/client-message-receipt.ts new file mode 100644 index 00000000..80146195 --- /dev/null +++ b/packages/protocol/src/msg-push/client-message-receipt.ts @@ -0,0 +1,157 @@ +import { CLIENT_MESSAGE_RECEIPT_CMD, type PacketInfo } from '@snowluma/common/protocol-types'; +import type { PacketSender } from '@snowluma/common/packet-sender'; +import { isRealUin } from '@snowluma/common/uin'; +import type { SendMessageResponse } from '@snowluma/proto-defs/action'; +import type { SsoGetC2cMsg, SsoGetC2cMsgResponse } from '@snowluma/proto-defs/get-c2c-msg'; +import { protobuf_decode, protobuf_encode } from '@snowluma/proton'; +import type { FriendMessage } from '../events'; +import type { IdentityService } from '../identity-service'; +import { classifyMessageSurvival } from './blank-filter'; +import { buildContextFromMessage } from './context'; +import { decodeFriendMessage } from './decoders/friend-message'; +import { PkgType } from './enums'; +import { SSO_GET_C2C_MSG_CMD } from './fetch-c2c-history'; +import { requirePacketResponse } from './packet-response'; + +export interface ClientMessageReceipt { + readonly peerUid: string; + readonly clientSequence: number; + readonly sequence: number; + readonly timestamp: number; +} + +export function decodeClientMessageReceipt(pkt: PacketInfo, selfUin: string): ClientMessageReceipt { + if ( + pkt.source !== 'client-message' || + pkt.serviceCmd !== CLIENT_MESSAGE_RECEIPT_CMD || + !isRealUin(selfUin) || + pkt.uin !== selfUin + ) { + throw new Error('Client message receipt has an invalid source or account'); + } + + if (pkt.retCode !== 0) { + throw new Error(`Client message receipt is unavailable (status=${pkt.retCode})`); + } + + const body = Buffer.from(pkt.body); + + if (body.length < 12 || body.length > 8 + 128 + 65536) { + throw new Error('Client message receipt has an invalid length'); + } + + const peerLength = body[1]!; + const ackOffset = 8 + peerLength; + + if ( + body[0] !== 1 || + body.readUInt16LE(2) !== 0 || + peerLength < 3 || + peerLength > 128 || + ackOffset >= body.length || + body.length - ackOffset > 65536 + ) { + throw new Error('Client message receipt has an invalid envelope'); + } + + const peerUid = body.subarray(8, ackOffset).toString('utf8'); + const clientSequence = body.readUInt32LE(4); + + if (!/^u_[A-Za-z0-9_-]+$/.test(peerUid) || clientSequence === 0) { + throw new Error('Client message receipt has an invalid message identity'); + } + + const ack = protobuf_decode(body.subarray(ackOffset)); + + if ( + (ack.result ?? 0) !== 0 || + (ack.reviewResult ?? 0) !== 0 || + (ack.serverResult?.code ?? 0) !== 0 + ) { + throw new Error('Client message was not accepted'); + } + + const sequence = ack.privateSequence; + const timestamp = ack.timestamp1; + + if (!sequence || !timestamp || (ack.groupSequence ?? 0) !== 0) { + throw new Error('Client message receipt has no valid private message confirmation'); + } + + return { peerUid, clientSequence, sequence, timestamp }; +} + +export async function fetchClientMessage( + sender: PacketSender, + identity: IdentityService, + receipt: ClientMessageReceipt, +): Promise { + const request = protobuf_encode({ + friendUid: receipt.peerUid, + startSequence: receipt.sequence, + endSequence: receipt.sequence, + }); + const result = await sender.sendPacket(SSO_GET_C2C_MSG_CMD, Buffer.from(request)); + const response = protobuf_decode( + requirePacketResponse(result, 'Client message lookup'), + ); + + if (response.friendUid && response.friendUid !== receipt.peerUid) { + throw new Error('Client message lookup returned another conversation'); + } + + if (response.messages?.length !== 1) { + throw new Error( + `Client message lookup expected one message, received ${response.messages?.length ?? 0}`, + ); + } + + const message = response.messages[0]!; + const head = message.contentHead; + const route = message.responseHead; + const selfUin = Number(identity.uin); + + if ( + !head || + !route || + route.fromUin !== selfUin || + route.toUid !== receipt.peerUid || + !route.toUin || + route.grp != null || + head.ntMsgSeq !== receipt.sequence || + head.sequence !== receipt.clientSequence || + head.timestamp !== receipt.timestamp + ) { + throw new Error('Client message lookup did not match the confirmed message'); + } + + if ( + ![ + PkgType.PrivateMessage, + PkgType.PrivateRecordMessage, + PkgType.PrivateFileMessage, + PkgType.ForwardFakePrivateMessage, + ].includes(head.msgType ?? 0) + ) { + throw new Error('Client message lookup returned an unsupported message type'); + } + + const context = buildContextFromMessage(message, selfUin, identity); + + if (!context) { + throw new Error('Client message lookup has no message context'); + } + + const events = decodeFriendMessage(context); + const event = events[0]; + + if ( + events.length !== 1 || + event?.kind !== 'friend_message' || + classifyMessageSurvival(context.head, event.elements, context.body) !== 'keep' + ) { + throw new Error('Client message lookup has no supported message content'); + } + + return event; +} diff --git a/packages/protocol/src/msg-push/index.ts b/packages/protocol/src/msg-push/index.ts index 336cbb81..df6c8255 100644 --- a/packages/protocol/src/msg-push/index.ts +++ b/packages/protocol/src/msg-push/index.ts @@ -26,6 +26,7 @@ import { deriveSysMsgDedupIdentity, SysMsgDedup } from './sysmsg-dedup'; export { deriveSysMsgDedupIdentity, SysMsgDedup } from './sysmsg-dedup'; export type { SysMsgDedupIdentity } from './sysmsg-dedup'; +export { decodeClientMessageReceipt, fetchClientMessage } from './client-message-receipt'; export { SSO_GET_GROUP_MSG_CMD, fetchGroupMessageRange } from './fetch-group-history'; export { diff --git a/packages/protocol/src/packet-pipeline.ts b/packages/protocol/src/packet-pipeline.ts index b19093b1..5556d366 100644 --- a/packages/protocol/src/packet-pipeline.ts +++ b/packages/protocol/src/packet-pipeline.ts @@ -85,8 +85,18 @@ export class IncomingPacketPipeline { } process(pkt: PacketInfo): Promise { + return this.processWithHandlers(pkt, this.cmdHandlers_.get(pkt.serviceCmd)); + } + + processEvents(pkt: PacketInfo, events: readonly QQEventVariant[]): Promise { + return this.processWithHandlers(pkt, [() => [...events]]); + } + + private processWithHandlers( + pkt: PacketInfo, + handlers: readonly CmdParser[] | undefined, + ): Promise { const startedAt = Date.now(); - const handlers = this.cmdHandlers_.get(pkt.serviceCmd); if (!handlers) { this.packetLog.trace(() => [ diff --git a/packages/protocol/tests/msg-push/client-message-receipt.test.ts b/packages/protocol/tests/msg-push/client-message-receipt.test.ts new file mode 100644 index 00000000..124d2607 --- /dev/null +++ b/packages/protocol/tests/msg-push/client-message-receipt.test.ts @@ -0,0 +1,292 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { CLIENT_MESSAGE_RECEIPT_CMD, type PacketInfo } from '@snowluma/common/protocol-types'; +import type { PacketSender, SendPacketResult } from '@snowluma/common/packet-sender'; +import type { SendMessageResponse } from '@snowluma/proto-defs/action'; +import type { SsoGetC2cMsg, SsoGetC2cMsgResponse } from '@snowluma/proto-defs/get-c2c-msg'; +import type { PushMsgBody } from '@snowluma/proto-defs/message'; +import { protobuf_decode, protobuf_encode } from '@snowluma/proton'; +import { IdentityService } from '../../src/identity-service'; +import { + decodeClientMessageReceipt, + fetchClientMessage, +} from '../../src/msg-push/client-message-receipt'; +import { SSO_GET_C2C_MSG_CMD } from '../../src/msg-push/fetch-c2c-history'; + +const self = '10001'; +const peer = 'u_test_peer'; +const time = 1_700_000_000; +const identities: IdentityService[] = []; + +afterEach(async () => { + await Promise.all(identities.splice(0).map((identity) => identity.close())); +}); + +function packet(ack: SendMessageResponse = { privateSequence: 87, timestamp1: time }): PacketInfo { + const uid = Buffer.from(peer); + const header = Buffer.alloc(8); + header[0] = 1; + header[1] = uid.length; + header.writeUInt32LE(9, 4); + return { + pid: 1234, + uin: self, + seqId: 42, + retCode: 0, + fromClient: false, + source: 'client-message', + serviceCmd: CLIENT_MESSAGE_RECEIPT_CMD, + body: Buffer.concat([header, uid, protobuf_encode(ack)]), + }; +} + +function message(): PushMsgBody { + return { + responseHead: { fromUin: Number(self), fromUid: 'u_test_self', toUin: 20002, toUid: peer }, + contentHead: { msgType: 166, sequence: 9, ntMsgSeq: 87, timestamp: time, msgId: 123456 }, + body: { richText: { elems: [{ text: { str: 'hello' } }, { face: { index: 1 } }] } }, + }; +} + +function setup(messages = [message()], friendUid = peer) { + const identity = IdentityService.memory(self); + identities.push(identity); + const sendPacket = vi.fn(async () => ({ + success: true, + gotResponse: true, + errorCode: 0, + errorMessage: '', + responseData: Buffer.from(protobuf_encode({ friendUid, messages })), + })); + const receipt = decodeClientMessageReceipt(packet(), self); + return { + sendPacket, + identity, + receipt, + run: () => fetchClientMessage({ sendPacket }, identity, receipt), + }; +} + +describe('client message receipts', () => { + it('decodes a confirmation and retrieves exactly its conversation and sequence', async () => { + const ctx = setup(); + expect(ctx.receipt).toEqual({ + peerUid: peer, + clientSequence: 9, + sequence: 87, + timestamp: time, + }); + const event = await ctx.run(); + expect(ctx.sendPacket).toHaveBeenCalledOnce(); + const [cmd, body] = ctx.sendPacket.mock.calls[0]!; + expect(cmd).toBe(SSO_GET_C2C_MSG_CMD); + expect(protobuf_decode(body)).toMatchObject({ + friendUid: peer, + startSequence: 87, + endSequence: 87, + }); + expect(event).toMatchObject({ + kind: 'friend_message', + selfUin: 10001, + senderUin: 10001, + peerUin: 20002, + clientSeq: 9, + ntMsgSeq: 87, + sequenceAuthoritative: true, + time, + elements: [ + { type: 'text', text: 'hello' }, + { type: 'face', faceId: 1 }, + ], + }); + }); + + it.each([ + { source: undefined }, + { serviceCmd: 'MessageSvc.PbSendMsg' }, + { uin: '20002' }, + { retCode: -1 }, + ] satisfies Partial[])('rejects invalid provenance or status: %j', (patch) => { + expect(() => decodeClientMessageReceipt({ ...packet(), ...patch }, self)).toThrow(); + }); + + it.each([ + { result: 1 }, + { reviewResult: 1 }, + { serverResult: { code: -1 } }, + { privateSequence: 0 }, + { timestamp1: 0 }, + { groupSequence: 5 }, + ] satisfies SendMessageResponse[])( + 'rejects an unsuccessful or incomplete confirmation: %j', + (patch) => { + expect(() => + decodeClientMessageReceipt( + packet({ privateSequence: 87, timestamp1: time, ...patch }), + self, + ), + ).toThrow(); + }, + ); + + it.each([ + [ + 'version', + (body: Buffer) => { + body[0] = 2; + }, + ], + [ + 'reserved', + (body: Buffer) => { + body[2] = 1; + }, + ], + [ + 'uid length', + (body: Buffer) => { + body[1] = 129; + }, + ], + [ + 'uid', + (body: Buffer) => { + body[8] = 0xff; + }, + ], + [ + 'client sequence', + (body: Buffer) => { + body.writeUInt32LE(0, 4); + }, + ], + ] as const)('rejects a malformed %s', (_name, mutate) => { + const pkt = packet(); + const body = Buffer.from(pkt.body); + mutate(body); + expect(() => decodeClientMessageReceipt({ ...pkt, body }, self)).toThrow(); + }); + + it('rejects truncated and excessive bodies', () => { + for (const body of [ + new Uint8Array(), + packet().body.subarray(0, 8 + peer.length), + new Uint8Array(8 + 128 + 65537), + ]) { + expect(() => decodeClientMessageReceipt({ ...packet(), body }, self)).toThrow(); + } + }); + + it.each([ + [ + 'sender', + (msg: PushMsgBody) => { + msg.responseHead!.fromUin = 20002; + }, + ], + [ + 'recipient', + (msg: PushMsgBody) => { + msg.responseHead!.toUid = 'u_other'; + }, + ], + [ + 'missing recipient', + (msg: PushMsgBody) => { + delete msg.responseHead!.toUin; + }, + ], + [ + 'client sequence', + (msg: PushMsgBody) => { + msg.contentHead!.sequence = 10; + }, + ], + [ + 'server sequence', + (msg: PushMsgBody) => { + msg.contentHead!.ntMsgSeq = 88; + }, + ], + [ + 'timestamp', + (msg: PushMsgBody) => { + msg.contentHead!.timestamp = time + 1; + }, + ], + [ + 'group', + (msg: PushMsgBody) => { + msg.responseHead!.grp = { groupCode: 30003 }; + }, + ], + [ + 'type', + (msg: PushMsgBody) => { + msg.contentHead!.msgType = 82; + }, + ], + [ + 'empty content', + (msg: PushMsgBody) => { + msg.body = {}; + }, + ], + [ + 'unknown content', + (msg: PushMsgBody) => { + msg.body = { richText: { elems: [{ commonElem: { serviceType: 999 } }] } }; + }, + ], + [ + 'control', + (msg: PushMsgBody) => { + msg.contentHead!.c2cCmd = 1; + }, + ], + ] as const)('rejects an incorrect %s in the lookup', async (_name, mutate) => { + const msg = message(); + mutate(msg); + await expect(setup([msg]).run()).rejects.toThrow(); + }); + + it('rejects missing, ambiguous and cross-conversation results', async () => { + await expect(setup([]).run()).rejects.toThrow('expected one'); + await expect(setup([message(), message()]).run()).rejects.toThrow('expected one'); + await expect(setup([message()], 'u_other').run()).rejects.toThrow('conversation'); + }); + + it('preserves distinct identities for equal text', async () => { + const first = setup(); + const msg = message(); + msg.contentHead = { ...msg.contentHead, sequence: 10, ntMsgSeq: 88 }; + const second = setup([msg]); + const event = await fetchClientMessage(second, second.identity, { + ...second.receipt, + clientSequence: 10, + sequence: 88, + }); + expect((await first.run()).ntMsgSeq).toBe(87); + expect(event.ntMsgSeq).toBe(88); + expect(event.elements).toEqual((await first.run()).elements); + }); + + it.each([ + { success: false, gotResponse: false, responseData: null }, + { success: true, gotResponse: false, responseData: null }, + { success: true, gotResponse: true, responseData: Buffer.alloc(0) }, + ] satisfies Partial[])( + 'surfaces unsuccessful lookup transport: %j', + async (result) => { + const ctx = setup(); + ctx.sendPacket.mockResolvedValue({ + success: true, + gotResponse: true, + errorCode: 0, + errorMessage: '', + responseData: null, + ...result, + }); + await expect(ctx.run()).rejects.toThrow('Client message lookup'); + }, + ); +});