From 464ff7398356d69f7c73923778e6cc3eed033db2 Mon Sep 17 00:00:00 2001 From: std-external <201375710+std-external@users.noreply.github.com> Date: Sun, 18 Oct 2026 14:27:26 +0800 Subject: [PATCH] qqnt: replace unstorable text units where the wire data arrives QQNT hands the relay text QQ copied from a person: a group title that is a contact's remark, a sender's nickname and remark, the message body, a forward title, a request note. PostgreSQL stores neither a NUL (`U+0000`) nor a UTF-16 surrogate without its partner in a `text`, `json` or `jsonb` column, and the event checkpoint advances only after the handler commits, so a single such unit failed the whole event and every later event replayed behind it forever. Clean exactly the fields that carry that text, once the wire data has been received and before any other step reads it: - `sanitize.ts` names the fields per wire object (message sender name/alias, text and markdown bodies, forward title/preview, card title/description, sticker title, media name/transcript, service notice wording and the names it carries, conversation title, request requester/group name and note, profile name/alias/signature). Ids, peer ids, URLs, MIME types, locators, enums and entity offsets are protocol values and stay untouched. - The event stream cleans each event at reception, before its filters, mapping and dispatch; the conversation/message/request mappers clean whatever a history, dialog, search or request read returns, and the profile reads clean a user's name and signature before they are persisted. Each unstorable code unit becomes `.` rather than disappearing, which keeps the replaced string exactly as long as it was: a text part's entities measure `offset`/`length` in code units of that text, so every mention range QQ sent still selects the same text and nothing has to be re-measured. A valid surrogate pair (an emoji) is one code point and is left alone. Tests: - `sanitize.test.ts` covers each named field, the length guarantee, a valid pair surviving, and a mention range still selecting `@alice` after a NUL. - `platform.test.ts` maps a message, a group title, a request and a profile with unstorable units and checks the mention entity slices. - `message-order.e2e.test.ts` replays a NUL message and a NUL group-join request over the WebSocket and asserts the persisted conversation title, user name, message text, mention range, sender remark and request note, plus the checkpoint reaching the second event. --- packages/platform-crossgram/src/index.ts | 69 +++--- .../src/message-order.e2e.test.ts | 138 ++++++++++++ .../platform-crossgram/src/platform.test.ts | 51 +++++ .../platform-crossgram/src/sanitize.test.ts | 204 ++++++++++++++++++ packages/platform-crossgram/src/sanitize.ts | 148 +++++++++++++ 5 files changed, 584 insertions(+), 26 deletions(-) create mode 100644 packages/platform-crossgram/src/sanitize.test.ts create mode 100644 packages/platform-crossgram/src/sanitize.ts diff --git a/packages/platform-crossgram/src/index.ts b/packages/platform-crossgram/src/index.ts index a4fcb738..a7ba62e0 100644 --- a/packages/platform-crossgram/src/index.ts +++ b/packages/platform-crossgram/src/index.ts @@ -29,6 +29,9 @@ import { defineQQMediaPreviewModel, mediaPreviewKey, QQMediaPreviewer } from './ import { extractRemoteVideoFrame, type RemoteVideoFrameExtractor } from './video-frame.js' import { migrateLegacyQQMessageMedia } from './raw-media-migration.js' import { migrateLegacyQQGroupAliasUsers } from './user-name-migration.js' +import { + sanitizeWireConversation, sanitizeWireEvent, sanitizeWireMessage, sanitizeWireRequest, sanitizeWireUser, +} from './sanitize.js' import type { QQMediaLocator, QQStickerReference, WireCallSignalEvent, WireCard, WireConversation, WireEvent, WireGroupFilePage, WireMedia, WireMessage, WireMultiForwardLocator, WireNativeAvsdkEvent, WireReactionState, WireRequest, WireTextPart, @@ -309,8 +312,9 @@ export class QQNTPlatform implements IMPlatform { || status.protocolVersion < MIN_PROTOCOL_VERSION) { throw new Error(`QQNT bridge protocol ${status.protocolVersion} is unsupported; protocol ${MIN_PROTOCOL_VERSION} or newer is required`) } - const user = await this.client.getUser(userId) - if (!user) throw new Error(`QQNT current user is unavailable: ${userId}`) + const profile = await this.client.getUser(userId) + if (!profile) throw new Error(`QQNT current user is unavailable: ${userId}`) + const user = sanitizeWireUser(profile) const qq = typeof status.selfUin === 'string' && /^\d+$/.test(status.selfUin) ? status.selfUin : typeof user.numericId === 'string' && /^\d+$/.test(user.numericId) @@ -418,7 +422,11 @@ export class QQNTPlatform implements IMPlatform { platformSessionId, attempt, this.client.webSocketEndpoint, lastEventId ?? '', ) try { - await this.client.subscribe(async (event, eventId) => { + await this.client.subscribe(async (rawEvent, eventId) => { + // Cleaned before any step of the stream reads the event: the + // checkpoint persists the event only after the handler commits, so + // one string PostgreSQL cannot store would stall every later event. + const event = sanitizeWireEvent(rawEvent) consecutiveStreamFailures = 0 try { if (event.type === 'native-avsdk') { @@ -837,14 +845,17 @@ export class QQNTPlatform implements IMPlatform { async getContacts(_session: PlatformSession, query: IMPageQuery = {}): Promise> { const page = await this.client.getContacts({ cursor: query.cursor, limit: query.limit }) return { - users: page.users.map((user) => ({ - id: user.id, - firstName: user.name, - username: user.numericId, - about: typeof user.signature === 'string' ? user.signature : undefined, - avatar: user.avatar ? mapMedia(user.avatar) : undefined, - metadata: qqUserMetadata(user), - })), + users: page.users.map((wireUser) => { + const user = sanitizeWireUser(wireUser) + return { + id: user.id, + firstName: user.name, + username: user.numericId, + about: typeof user.signature === 'string' ? user.signature : undefined, + avatar: user.avatar ? mapMedia(user.avatar) : undefined, + metadata: qqUserMetadata(user), + } + }), nextCursor: page.nextCursor, } } @@ -997,8 +1008,9 @@ export class QQNTPlatform implements IMPlatform { } async getUser(_session: PlatformSession, userId: string): Promise | null> { - const user = await this.client.getUser(userId) - if (!user) return null + const wireUser = await this.client.getUser(userId) + if (!wireUser) return null + const user = sanitizeWireUser(wireUser) return { id: user.id, firstName: user.name, @@ -1089,18 +1101,21 @@ export class QQNTPlatform implements IMPlatform { if (isFilteredConversationId(conversation.id)) return { members: [], total: 0 } const page = await this.client.getMembers(conversation.id, { cursor: query.cursor, limit: query.limit }) return { - members: page.members.map((member): IMConversationMember => ({ - user: { - id: member.user.id, - firstName: member.user.name, - username: member.user.numericId, - avatar: member.user.avatar ? mapMedia(member.user.avatar) : undefined, - metadata: { ...qqUserMetadata(member.user), qqName: member.user.name }, - }, - role: member.role, - permissions: permissions(member.role), - title: member.user.alias?.trim() || undefined, - })), + members: page.members.map((member): IMConversationMember => { + const user = sanitizeWireUser(member.user) + return { + user: { + id: user.id, + firstName: user.name, + username: user.numericId, + avatar: user.avatar ? mapMedia(user.avatar) : undefined, + metadata: { ...qqUserMetadata(user), qqName: user.name }, + }, + role: member.role, + permissions: permissions(member.role), + title: user.alias?.trim() || undefined, + } + }), total: page.total, nextCursor: page.nextCursor, } @@ -2238,6 +2253,7 @@ export class QQNTPlatform implements IMPlatform { } private mapConversation(input: WireConversation, session?: PlatformSession): IMConversation { + input = sanitizeWireConversation(input) const id = session && this.savedMessagesConversationIds.get(session.platformSessionId) === input.id ? session.userId : input.id @@ -2337,7 +2353,7 @@ export class QQNTPlatform implements IMPlatform { } private mapMessage(input: WireMessage, conversationId?: string): IMMessage { - input = this.splitOutgoingMessages.get(input.id) ?? input + input = sanitizeWireMessage(this.splitOutgoingMessages.get(input.id) ?? input) const message = mapMessage( conversationId ? { ...input, conversationId } : input, this.reactionCatalog, this.stickerProviderId, @@ -2597,6 +2613,7 @@ function mapConversation(input: WireConversation): IMConversation { + input = sanitizeWireRequest(input) return { id: input.id, kind: input.kind, diff --git a/packages/platform-crossgram/src/message-order.e2e.test.ts b/packages/platform-crossgram/src/message-order.e2e.test.ts index 7a8ab1a9..ff23a8f0 100644 --- a/packages/platform-crossgram/src/message-order.e2e.test.ts +++ b/packages/platform-crossgram/src/message-order.e2e.test.ts @@ -2601,6 +2601,144 @@ describe('QQNT join notice E2E', () => { }) }) +describe('QQNT unstorable text E2E', () => { + it('persists a NUL nickname, remark, body and request note, and keeps the checkpoint moving', async () => { + const ctx = new Context() + const fibers = [ + ctx.plugin(Database), + ctx.plugin(SQLiteDriver, { path: ':memory:' }), + ] + await Promise.all(fibers) + await new Promise((resolve) => setTimeout(resolve, 25)) + defineModels(ctx) + defineQQNTEventCheckpointModel(ctx) + await ctx.database.prepared() + disposals.push(async () => { + for (const fiber of fibers.reverse()) await Promise.resolve((fiber as any).dispose?.()) + }) + + const checkpointSession = { ...session, platformSessionId: 'qqnt-unstorable-text-e2e' } + const conversation = { + id: 'unstorable-group', kind: 'group' as const, title: 'Gr\u0000oup', + peerUid: 'unstorable-group', peerUin: '43', chatType: 2 as const, + } + const webSocketServer = new WebSocketServer({ noServer: true }) + const server = createServer() + server.on('upgrade', (request, socket, head) => { + webSocketServer.handleUpgrade(request, socket, head, (webSocket) => { + webSocketServer.emit('connection', webSocket, request) + }) + }) + webSocketServer.on('connection', (webSocket) => { + // QQ hands over a NUL in a group title, a nickname, a remark and a body, + // and a lone surrogate in the remark: PostgreSQL stores none of them, so + // before the fix the whole event failed and this checkpoint never moved. + webSocket.send(JSON.stringify({ + id: '7001', + event: { + type: 'message', conversation, + message: { + id: 'unstorable-message', conversationId: conversation.id, senderId: 'peer', + timestamp: 1_787_215_100, outgoing: false, + sender: { id: 'peer', name: 'Bad\u0000Name', alias: 're\uD800mark' }, + parts: [{ + type: 'text', text: '@alice\u0000 hello', + entities: [{ type: 'mention', offset: 0, length: 6, userId: 'alice' }], + }], + }, + }, + })) + webSocket.send(JSON.stringify({ + id: '7002', + event: { + type: 'request', + request: { + id: 'unstorable-request', kind: 'group-join', status: 'pending', + requester: { id: 'requester', name: 'Frie\u0000nd' }, + group: { id: conversation.id, name: 'Gr\u0000oup' }, + message: 'let me in\u0000', + timestamp: 1_787_215_101, + }, + }, + })) + }) + server.listen(0, '127.0.0.1') + await once(server, 'listening') + const address = server.address() + if (!address || typeof address === 'string') throw new Error('missing unstorable text test address') + disposals.push(async () => { + for (const client of webSocketServer.clients) client.terminate() + webSocketServer.close() + if (!server.listening) return + const closed = new Promise((resolve, reject) => { + server.close((error) => error ? reject(error) : resolve()) + }) + server.closeAllConnections() + await closed + }) + + const platform = new QQNTPlatform({ + endpoint: 'http://127.0.0.1:1/v1', + webSocketEndpoint: `ws://127.0.0.1:${address.port}/events`, + }, 'qqnt:stickers', undefined, undefined, ctx.database) + platform.client.getReactionCatalog = vi.fn(async () => ({ + available: [], reactions: [], maxSelected: 20, + })) + platform.client.getDialogs = vi.fn(async () => ({ conversations: [] })) + const store = new MessageStore(ctx.database) + const handled: string[] = [] + const done = Promise.withResolvers() + const unsubscribe = await platform.subscribe(checkpointSession, async (event) => { + if (event.type === 'message') await store.ingest(checkpointSession, event.conversation, event.message) + else if (event.type === 'request') await store.ingestRequest(checkpointSession, event.request) + else return + handled.push(event.type) + if (handled.length === 2) done.resolve() + }) + try { + await Promise.race([ + done.promise, + new Promise((_, reject) => setTimeout(() => reject(new Error('unstorable text E2E timed out')), 5_000)), + ]) + expect(handled).toEqual(['message', 'request']) + // The reported symptom: the checkpoint stops before the event that cannot + // be persisted, and every later event waits behind it. + await vi.waitFor(async () => expect(await ctx.database.get( + 'mtproto_qqnt_event_checkpoint', { platformSessionId: checkpointSession.platformSessionId }, + )).toMatchObject([{ lastEventId: '7002' }])) + await expect(ctx.database.get('mtproto_im_conversation', { + platformSessionId: checkpointSession.platformSessionId, platformConversationId: conversation.id, + })).resolves.toMatchObject([{ title: 'Gr.oup' }]) + await expect(ctx.database.get('mtproto_im_user', { + platformId: 'qqnt', platformUserId: 'peer', + })).resolves.toMatchObject([{ firstName: 'Bad.Name' }]) + const [message] = await ctx.database.get('mtproto_im_message', { + platformSessionId: checkpointSession.platformSessionId, + primaryPlatformMessageId: 'unstorable-message', + }) + expect(message.text).toBe('@alice. hello') + // The mention still covers the same six code units it covered in QQ's text. + expect(message.content).toMatchObject({ + parts: [{ + type: 'text', text: '@alice. hello', + entities: [{ type: 'mention', offset: 0, length: 6, userId: 'alice' }], + }], + }) + expect(message.metadata).toMatchObject({ __mtprotoRelaySenderTitle: 're.mark' }) + const [requestRow] = await ctx.database.get('mtproto_im_request', { + platformSessionId: checkpointSession.platformSessionId, platformRequestId: 'unstorable-request', + }) + expect(requestRow.request).toMatchObject({ + message: 'let me in.', + requester: { id: 'requester', firstName: 'Frie.nd' }, + group: { id: conversation.id, title: 'Gr.oup' }, + }) + } finally { + await unsubscribe() + } + }) +}) + async function collect(source: AsyncIterable): Promise { const chunks: Buffer[] = [] for await (const chunk of source) chunks.push(Buffer.from(chunk)) diff --git a/packages/platform-crossgram/src/platform.test.ts b/packages/platform-crossgram/src/platform.test.ts index a5d42173..848d72c9 100644 --- a/packages/platform-crossgram/src/platform.test.ts +++ b/packages/platform-crossgram/src/platform.test.ts @@ -72,6 +72,57 @@ describe('QQNTPlatform mapping', () => { ]) }) + it('keeps a mention range and the sender text when QQ hands over a NUL', async () => { + const platform = new QQNTPlatform() + platform.client.getHistory = vi.fn(async () => ({ + messages: [{ + id: 'nul-message', conversationId: 'group', senderId: 'peer', timestamp: 1, outgoing: false, + sender: { id: 'peer', name: 'Bad\u0000Name', alias: 're\u0000mark' }, + parts: [{ + type: 'text' as const, text: '@alice\u0000 hello', + entities: [{ type: 'mention' as const, offset: 0, length: 6, userId: 'alice' }], + }], + }], + })) + + const [message] = (await platform.getHistory(session, { id: 'group' })).messages + expect(message.sender).toMatchObject({ firstName: 'Bad.Name' }) + expect((message.sender?.metadata as { qqName?: string }).qqName).toBe('Bad.Name') + expect(message.senderTitle).toBe('re.mark') + const part = message.content.parts[0] + expect(part).toMatchObject({ type: 'text', text: '@alice. hello' }) + expect(entitySlices(part as never)).toEqual(['mention:@alice']) + }) + + it('keeps a group title and a request note free of the units PostgreSQL rejects', async () => { + const platform = new QQNTPlatform() + platform.client.getConversation = vi.fn(async () => ({ + id: 'group', kind: 'group' as const, title: 'Gr\u0000oup\uD800', + peerUid: 'group', peerUin: '42', chatType: 2 as const, + })) + platform.client.getRequests = vi.fn(async () => ({ requests: [{ + id: 'request', kind: 'friend' as const, status: 'pending' as const, + requester: { id: 'peer', name: 'Frie\u0000nd' }, + group: { id: 'group', name: 'Gr\u0000oup' }, + message: 'let me in\u0000', + timestamp: 1, + }] })) + platform.client.getUser = vi.fn(async () => ({ + id: 'peer', name: 'Na\u0000me', signature: 'sig\uD800', + })) + + await expect(platform.getConversation(session, 'group')).resolves.toMatchObject({ title: 'Gr.oup.' }) + const [request] = (await platform.getRequests(session)).requests + expect(request).toMatchObject({ + requester: { firstName: 'Frie.nd' }, + group: { title: 'Gr.oup' }, + message: 'let me in.', + }) + await expect(platform.getUser(session, 'peer')).resolves.toMatchObject({ + firstName: 'Na.me', about: 'sig.', + }) + }) + it('maps QQNT ranged-read past EOF responses to an empty Telegram chunk', async () => { const platform = new QQNTPlatform() platform.client.downloadFile = vi.fn(async function* () { diff --git a/packages/platform-crossgram/src/sanitize.test.ts b/packages/platform-crossgram/src/sanitize.test.ts new file mode 100644 index 00000000..603d369f --- /dev/null +++ b/packages/platform-crossgram/src/sanitize.test.ts @@ -0,0 +1,204 @@ +import { describe, expect, it } from 'vitest' +import { + sanitizeWireConversation, sanitizeWireEvent, sanitizeWireMessage, sanitizeWireRequest, sanitizeWireText, +} from './sanitize.js' +import type { WireConversation, WireEvent, WireMessage, WireRequest, WireTextPart } from './protocol.js' + +const locator = { + messageId: 'message', elementId: 'element', chatType: 2 as const, + peerUid: 'group', kind: 'file' as const, fileName: 'a\u0000b.bin', +} + +const conversation: WireConversation = { + id: 'group', kind: 'group', title: 'Group', peerUid: 'group', peerUin: '42', chatType: 2, +} + +function textPart(message: WireMessage, index = 0): WireTextPart { + const part = message.parts[index] + if (part.type !== 'text') throw new Error('expected a text part') + return part +} + +describe('QQNT wire text sanitizing', () => { + it('replaces a NUL and a lone surrogate with one dot each and leaves valid text alone', () => { + expect(sanitizeWireText('a\u0000b')).toBe('a.b') + expect(sanitizeWireText('a\uD800b')).toBe('a.b') + expect(sanitizeWireText('a\uDC00b')).toBe('a.b') + expect(sanitizeWireText('\u0000😀')).toBe('.😀') + // A surrogate pair is one code point, not two unstorable surrogates. + expect(sanitizeWireText('😀')).toBe('😀') + expect(sanitizeWireText('plain')).toBe('plain') + }) + + it('keeps every replaced text exactly as long as it was', () => { + for (const text of ['\u0000', 'x\u0000y', 'pre\uD800', 'a\uDC00\u0000b', '😀\uD800😀']) { + expect(sanitizeWireText(text)).toHaveLength(text.length) + } + }) + + it('leaves a mention range pointing at the same text after a NUL in the body', () => { + const part = textPart(sanitizeWireMessage({ + id: 'message', conversationId: 'group', senderId: 'peer', timestamp: 1, outgoing: false, + parts: [{ + type: 'text', + text: '@alice\u0000 hello', + entities: [{ type: 'mention', offset: 0, length: 6, userId: 'alice' }], + }], + })) + expect(part).toMatchObject({ + text: '@alice. hello', + entities: [{ type: 'mention', offset: 0, length: 6, userId: 'alice' }], + }) + const entity = part.entities![0] + expect(part.text.slice(entity.offset, entity.offset + entity.length)).toBe('@alice') + }) + + it('cleans the nickname and remark of a message sender', () => { + const message = sanitizeWireMessage({ + id: 'message', conversationId: 'group', senderId: 'peer', timestamp: 1, outgoing: false, + sender: { id: 'peer', numericId: '10001', name: 'Bad\u0000Name', alias: 're\uD800mark' }, + parts: [], + }) + expect(message.sender).toMatchObject({ + id: 'peer', numericId: '10001', name: 'Bad.Name', alias: 're.mark', + }) + }) + + it('cleans the text a message carries, and nothing else', () => { + const parts: WireMessage['parts'] = [ + { type: 'text', text: 'bo\u0000dy', entities: [{ type: 'qq-face', offset: 0, length: 2, faceId: '1', faceType: 1 }] }, + { type: 'markdown', content: '# head\uD800line' }, + { type: 'multi-forward', title: 'chats\u0000', preview: 'pre\u0000view', locator: { + conversationId: 'group', rootMessageId: 'root', + } }, + { type: 'card', card: { + kind: 'link', title: 'a\u0000 title', description: 'de\u0000sc', + url: 'https://example.com/\u0000', thumbnailUrl: 'https://example.com/t.png', + } }, + { type: 'sticker', sticker: { + stickerId: 'sticker', title: 'pack\u0000title', format: 'static', mimeType: 'image/png', + reference: { kind: 'sysface', faceId: '1', faceType: 1, name: 'n', animated: true }, + } }, + { type: 'media', media: { + id: 'media', kind: 'file', name: 'a\u0000b.bin', mimeType: 'application/octet-stream', + transcript: 'said\uD800it', locator, + } }, + ] + const cleaned = sanitizeWireMessage({ + id: 'message', conversationId: 'group', senderId: 'peer', timestamp: 1, outgoing: false, + sender: { id: 'peer', name: 'Name' }, + parts, + serviceAction: { type: 'custom', text: 'joined\u0000the group' }, + }) + + expect(cleaned.parts[0]).toEqual({ + type: 'text', text: 'bo.dy', entities: [{ type: 'qq-face', offset: 0, length: 2, faceId: '1', faceType: 1 }], + }) + expect(cleaned.parts[1]).toEqual({ type: 'markdown', content: '# head.line' }) + expect(cleaned.parts[2]).toMatchObject({ title: 'chats.', preview: 'pre.view' }) + // A card's URL is a protocol value, not text QQ copied from a person. + expect(cleaned.parts[3]).toEqual({ + type: 'card', + card: { + kind: 'link', title: 'a. title', description: 'de.sc', + url: 'https://example.com/\u0000', thumbnailUrl: 'https://example.com/t.png', + }, + }) + expect(cleaned.parts[4]).toMatchObject({ sticker: { title: 'pack.title', mimeType: 'image/png' } }) + expect(cleaned.parts[5]).toMatchObject({ + media: { name: 'a.b.bin', transcript: 'said.it', mimeType: 'application/octet-stream', locator }, + }) + expect(cleaned.serviceAction).toEqual({ type: 'custom', text: 'joined.the group' }) + expect(cleaned.id).toBe('message') + }) + + it('cleans the names a join notice carries beside its wording', () => { + const cleaned = sanitizeWireMessage({ + id: 'message', conversationId: 'group', senderId: 'peer', timestamp: 1, outgoing: false, parts: [], + serviceAction: { + type: 'members-joined', text: 'A\u0000B joined', + members: [{ id: 'a', name: 'Al\u0000ice' }, { id: 'b', name: 'Bob' }], + actor: { id: 'a', name: 'Al\u0000ice' }, + }, + }) + expect(cleaned.serviceAction).toEqual({ + type: 'members-joined', text: 'A.B joined', + members: [{ id: 'a', name: 'Al.ice' }, { id: 'b', name: 'Bob' }], + actor: { id: 'a', name: 'Al.ice' }, + }) + }) + + it('cleans a conversation title, which is a group name or a contact remark', () => { + const cleaned = sanitizeWireConversation({ ...conversation, title: 'Gr\u0000oup\uD800' }) + expect(cleaned.title).toBe('Gr.oup.') + expect(cleaned.id).toBe('group') + expect(cleaned.chatType).toBe(2) + }) + + it('cleans a request note and both names it names', () => { + const request: WireRequest = { + id: 'request', kind: 'friend', status: 'pending', + requester: { id: 'peer', name: 'Frie\u0000nd' }, + group: { id: 'group', name: 'Gr\u0000oup' }, + message: 'let me in\u0000', + timestamp: 1, + } + expect(sanitizeWireRequest(request)).toMatchObject({ + id: 'request', kind: 'friend', status: 'pending', + requester: { id: 'peer', name: 'Frie.nd' }, + group: { id: 'group', name: 'Gr.oup' }, + message: 'let me in.', + }) + }) + + it('cleans every event kind before the stream reads it', () => { + const message: WireMessage = { + id: 'message', conversationId: 'group', senderId: 'peer', timestamp: 1, outgoing: false, + sender: { id: 'peer', name: 'Na\u0000me' }, + parts: [{ type: 'text', text: 'bo\u0000dy' }], + } + expect(sanitizeWireEvent({ + type: 'message', conversation: { ...conversation, title: 'Gr\u0000oup' }, message, + })).toMatchObject({ + conversation: { title: 'Gr.oup' }, + message: { sender: { name: 'Na.me' }, parts: [{ text: 'bo.dy' }] }, + }) + expect(sanitizeWireEvent({ + type: 'message-edit', eventId: 'edit', conversation, message, + })).toMatchObject({ message: { parts: [{ text: 'bo.dy' }] } }) + expect(sanitizeWireEvent({ + type: 'request', + request: { id: 'request', kind: 'group-join', status: 'pending', requester: { id: 'peer', name: 'Na\u0000me' }, message: 'no\u0000te' }, + })).toMatchObject({ request: { requester: { name: 'Na.me' }, message: 'no.te' } }) + expect(sanitizeWireEvent({ + type: 'voice-transcript', eventId: 'voice', conversation, messageId: 'm', transcript: 'sa\u0000id', timestamp: 1, + })).toMatchObject({ transcript: 'sa.id' }) + // An event that carries no text of its own is handed on untouched. + const deletion: WireEvent = { type: 'message-delete', eventId: 'delete', conversation, messageIds: ['m'], timestamp: 1 } + expect(sanitizeWireEvent(deletion)).toBe(deletion) + }) + + it('returns the event itself when nothing in it needed cleaning', () => { + const message: WireMessage = { + id: 'message', conversationId: 'group', senderId: 'peer', timestamp: 1, outgoing: false, + sender: { id: 'peer', name: 'Name' }, parts: [{ type: 'text', text: 'body' }], + } + const event: WireEvent = { type: 'message', conversation, message } + expect(sanitizeWireEvent(event)).toBe(event) + expect(sanitizeWireConversation(conversation)).toBe(conversation) + expect(sanitizeWireMessage(message)).toBe(message) + const request: WireRequest = { id: 'r', kind: 'friend', status: 'pending', requester: { id: 'p', name: 'N' } } + expect(sanitizeWireRequest(request)).toBe(request) + }) + + it('leaves an absent alias, request group and note absent', () => { + const message = sanitizeWireMessage({ + id: 'message', conversationId: 'group', senderId: 'peer', timestamp: 1, outgoing: false, + sender: { id: 'peer', name: 'Na\u0000me' }, parts: [], + }) + expect(message.sender).not.toHaveProperty('alias') + expect(sanitizeWireRequest({ + id: 'request', kind: 'friend', status: 'pending', requester: { id: 'peer', name: 'Na\u0000me' }, + })).not.toHaveProperty('group') + }) +}) diff --git a/packages/platform-crossgram/src/sanitize.ts b/packages/platform-crossgram/src/sanitize.ts new file mode 100644 index 00000000..50c12337 --- /dev/null +++ b/packages/platform-crossgram/src/sanitize.ts @@ -0,0 +1,148 @@ +import type { WireConversation, WireEvent, WireMessage, WireRequest, WireServiceAction } from './protocol.js' + +/** + * QQNT carries text QQ copied from a person: nicknames, remarks, message + * bodies, forward titles, service notice wording. PostgreSQL stores no NUL + * (`U+0000`) and no UTF-16 surrogate without its partner in a `text`, `json` + * or `jsonb` column, and an event is persisted as a whole: one such unit in + * one of those strings fails the event, the durable checkpoint never advances, + * and the same event replays forever while every later event waits behind it. + * + * The units are replaced where the wire data is received, before any other + * step reads it, and each one becomes a single `.` instead of disappearing: a + * text part's entities measure `offset`/`length` in code units of that text, + * so replacing one code unit with one code unit keeps every offset QQ sent + * valid and nothing has to be re-measured. + * + * Only the named fields below are cleaned, and each one is a string QQ copies + * from a person rather than a protocol value: ids, peer ids, URLs, MIME types, + * locators, enum values and entity offsets are left exactly as sent. + */ + +/** + * NUL, or a surrogate without its partner. A valid pair is one code point + * outside the surrogate range, so an emoji stays an emoji. + */ +const UNSTORABLE_UNIT = /\u0000|[\uD800-\uDFFF]/gu + +/** Replace every code unit PostgreSQL cannot store, preserving the length. */ +export function sanitizeWireText(text: string): string { + return text.replace(UNSTORABLE_UNIT, '.') +} + +/** Clean the named string fields of `value`, leaving every other field alone. */ +function sanitizeFields(value: T, keys: readonly (keyof T & string)[]): T { + let output: Record | undefined + for (const key of keys) { + const item = value[key] + if (typeof item !== 'string') continue + const clean = sanitizeWireText(item) + if (clean === item) continue + output ??= { ...value } as Record + output[key] = clean + } + return (output as T | undefined) ?? value +} + +type WirePart = WireMessage['parts'][number] + +function sanitizeWirePart(part: WirePart): WirePart { + switch (part.type) { + case 'text': + // Entities keep the offsets and lengths QQ measured on this same text, + // which the one-code-unit-for-one-code-unit replacement leaves correct. + return sanitizeFields(part, ['text']) + case 'markdown': + return sanitizeFields(part, ['content']) + case 'multi-forward': + return sanitizeFields(part, ['title', 'preview']) + case 'card': { + const card = sanitizeFields(part.card, ['title', 'description']) + return card === part.card ? part : { ...part, card } + } + case 'sticker': { + const sticker = sanitizeFields(part.sticker, ['title']) + return sticker === part.sticker ? part : { ...part, sticker } + } + case 'media': { + const media = sanitizeFields(part.media, ['name', 'transcript']) + return media === part.media ? part : { ...part, media } + } + // A keyboard is a bot's protocol payload, not text a person typed. + case 'inline-keyboard': + return part + } +} + +function sanitizeServiceAction(action: WireServiceAction): WireServiceAction { + if (action.type === 'custom') return sanitizeFields(action, ['text']) + if (action.type !== 'members-joined') return action + const named = sanitizeFields(action, ['text']) + const members = action.members.map((member) => sanitizeFields(member, ['name'])) + const actor = action.actor && sanitizeFields(action.actor, ['name']) + const membersChanged = members.some((member, index) => member !== action.members[index]) + if (named === action && !membersChanged && actor === action.actor) return action + return { + ...named, + ...(membersChanged ? { members } : {}), + ...(actor === action.actor ? {} : { actor }), + } as WireServiceAction +} + +/** A message: the sender's nickname and remark, and everything the message carries as text. */ +export function sanitizeWireMessage(message: WireMessage): WireMessage { + const sender = message.sender && sanitizeFields(message.sender, ['name', 'alias']) + const parts = message.parts.map(sanitizeWirePart) + const partsChanged = parts.some((part, index) => part !== message.parts[index]) + const serviceAction = message.serviceAction && sanitizeServiceAction(message.serviceAction) + if (sender === message.sender && !partsChanged && serviceAction === message.serviceAction) return message + return { + ...message, + sender, + ...(partsChanged ? { parts } : {}), + ...(serviceAction === message.serviceAction ? {} : { serviceAction }), + } +} + +/** A conversation: its title, which is a group's name or a contact's remark. */ +export function sanitizeWireConversation(conversation: WireConversation): WireConversation { + return sanitizeFields(conversation, ['title']) +} + +/** A friend or group request: the requester's name, the group's name and the note they wrote. */ +export function sanitizeWireRequest(request: WireRequest): WireRequest { + const requester = sanitizeFields(request.requester, ['name']) + const group = request.group && sanitizeFields(request.group, ['name']) + const message = request.message === undefined ? undefined : sanitizeWireText(request.message) + if (requester === request.requester && group === request.group && message === request.message) return request + return { + ...request, + requester, + ...(group === request.group ? {} : { group }), + ...(message === request.message ? {} : { message }), + } +} + +/** A profile: the nickname and signature carried by a contact, member or user lookup. */ +export function sanitizeWireUser(user: T): T { + return sanitizeFields(user, ['name', 'alias', 'signature']) +} + +/** Every event QQNT delivers, cleaned before the stream reads any field of it. */ +export function sanitizeWireEvent(event: Event): Event { + if (event.type === 'native-avsdk') return event + if (event.type === 'request') { + const request = sanitizeWireRequest(event.request) + return (request === event.request ? event : { ...event, request }) as Event + } + const conversation = sanitizeWireConversation(event.conversation) + let cleaned: Event = conversation === event.conversation ? event : { ...event, conversation } + if (cleaned.type === 'message' || cleaned.type === 'message-edit') { + const message = sanitizeWireMessage(cleaned.message) + if (message !== cleaned.message) cleaned = { ...cleaned, message } + } else if (cleaned.type === 'voice-transcript') { + const transcript = sanitizeWireText(cleaned.transcript) + if (transcript !== cleaned.transcript) cleaned = { ...cleaned, transcript } + } + return cleaned +}