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 +}