From 0043f8be214d92b937ae215c0f4c47953516870f Mon Sep 17 00:00:00 2001 From: Gustavo Quadri <87215048+gusquadri@users.noreply.github.com> Date: Sun, 7 Sep 2025 08:12:24 -0300 Subject: [PATCH] Add: LID support with lid-map and migrations of sessions (#1694) * fix: add lid-map logic * fix: lint and simplified version * fix: single device migration * fix: lid map discovery * fix: lint * fix: lid migration in socket connection * fix: lint * fix: decode-wa-message consistency * fix: yarn lock * fix: lint * fix: retry logic * fix: logger level --- package.json | 1 + src/Signal/libsignal.ts | 192 ++++++++++- src/Signal/lid-mapping.ts | 101 ++++++ src/Socket/messages-recv.ts | 39 +++ src/Socket/messages-send.ts | 601 +++++++++++++++++++++++++++++---- src/Socket/socket.ts | 19 ++ src/Types/Auth.ts | 1 + src/Types/Signal.ts | 18 + src/Utils/decode-wa-message.ts | 70 +++- yarn.lock | 3 +- 10 files changed, 968 insertions(+), 77 deletions(-) create mode 100644 src/Signal/lid-mapping.ts diff --git a/package.json b/package.json index 336110b9..aae23539 100644 --- a/package.json +++ b/package.json @@ -69,6 +69,7 @@ "jiti": "^2.4.2", "json": "^11.0.0", "link-preview-js": "^3.0.0", + "lru-cache": "^10.4.3", "open": "^8.4.2", "prettier": "^3.5.3", "protobufjs-cli": "^1.1.3", diff --git a/src/Signal/libsignal.ts b/src/Signal/libsignal.ts index e5e74d26..a1fd2acf 100644 --- a/src/Signal/libsignal.ts +++ b/src/Signal/libsignal.ts @@ -1,6 +1,8 @@ /* @ts-ignore */ import * as libsignal from 'libsignal' -import type { SignalAuthState } from '../Types' +/* @ts-ignore */ +import { LRUCache } from 'lru-cache' +import type { SignalAuthState, SignalKeyStoreWithTransaction } from '../Types' import type { SignalRepository } from '../Types/Signal' import { generateSignalPubKey } from '../Utils' import { jidDecode } from '../WABinary' @@ -8,10 +10,18 @@ import type { SenderKeyStore } from './Group/group_cipher' import { SenderKeyName } from './Group/sender-key-name' import { SenderKeyRecord } from './Group/sender-key-record' import { GroupCipher, GroupSessionBuilder, SenderKeyDistributionMessage } from './Group' +import { LIDMappingStore } from './lid-mapping' export function makeLibSignalRepository(auth: SignalAuthState): SignalRepository { - const storage: SenderKeyStore = signalStorage(auth) - return { + const lidMapping = new LIDMappingStore(auth.keys as SignalKeyStoreWithTransaction) + const storage = signalStorage(auth, lidMapping) + // Simple operation-level deduplication (5 minutes) + const recentMigrations = new LRUCache({ + max: 500, + ttl: 5 * 60 * 1000 + }) + + const repository: SignalRepository = { decryptGroupMessage({ group, authorJid, msg }) { const senderName = jidToSignalSenderKeyName(group, authorJid) const cipher = new GroupCipher(storage, senderName) @@ -58,8 +68,36 @@ export function makeLibSignalRepository(auth: SignalAuthState): SignalRepository return result }, + async encryptMessage({ jid, data }) { - const addr = jidToSignalProtocolAddress(jid) + // LID SINGLE SOURCE OF TRUTH: Always prefer LID when available + let encryptionJid = jid + + // Check for LID mapping and use it if session exists + if (jid.includes('@s.whatsapp.net')) { + const lidForPN = await lidMapping.getLIDForPN(jid) + if (lidForPN?.includes('@lid')) { + const lidAddr = jidToSignalProtocolAddress(lidForPN) + const { [lidAddr.toString()]: lidSession } = await auth.keys.get('session', [lidAddr.toString()]) + + if (lidSession) { + // LID session exists, use it + encryptionJid = lidForPN + } else { + // Try to migrate if PN session exists + const pnAddr = jidToSignalProtocolAddress(jid) + const { [pnAddr.toString()]: pnSession } = await auth.keys.get('session', [pnAddr.toString()]) + + if (pnSession) { + // Migrate PN to LID + await repository.migrateSession(jid, lidForPN) + encryptionJid = lidForPN + } + } + } + } + + const addr = jidToSignalProtocolAddress(encryptionJid) const cipher = new libsignal.SessionCipher(storage, addr) const { type: sigType, body } = await cipher.encrypt(data) @@ -91,26 +129,160 @@ export function makeLibSignalRepository(auth: SignalAuthState): SignalRepository }, jidToSignalProtocolAddress(jid) { return jidToSignalProtocolAddress(jid).toString() + }, + + async storeLIDPNMapping(lid: string, pn: string) { + await lidMapping.storeLIDPNMapping(lid, pn) + }, + + getLIDMappingStore() { + return lidMapping + }, + + async validateSession(jid: string) { + try { + const addr = jidToSignalProtocolAddress(jid) + const session = await storage.loadSession(addr.toString()) + + if (!session) { + return { exists: false, reason: 'no session' } + } + + if (!session.haveOpenSession()) { + return { exists: false, reason: 'no open session' } + } + + return { exists: true } + } catch (error) { + return { exists: false, reason: 'validation error' } + } + }, + + async deleteSession(jid: string) { + const addr = jidToSignalProtocolAddress(jid) + + return (auth.keys as SignalKeyStoreWithTransaction).transaction(async () => { + await auth.keys.set({ session: { [addr.toString()]: null } }) + }) + }, + + async migrateSession(fromJid: string, toJid: string) { + // Only migrate PN → LID + if (!fromJid.includes('@s.whatsapp.net') || !toJid.includes('@lid')) { + return + } + + const fromDecoded = jidDecode(fromJid) + const toDecoded = jidDecode(toJid) + if (!fromDecoded || !toDecoded) return + + const deviceId = fromDecoded.device || 0 + const migrationKey = `${fromDecoded.user}.${deviceId}→${toDecoded.user}.${deviceId}` + + // Check if recently migrated (5 min window) + if (recentMigrations.has(migrationKey)) { + return + } + + // Check if LID session already exists + const lidAddr = jidToSignalProtocolAddress(toJid) + const { [lidAddr.toString()]: lidExists } = await auth.keys.get('session', [lidAddr.toString()]) + if (lidExists) { + recentMigrations.set(migrationKey, true) + return + } + + return (auth.keys as SignalKeyStoreWithTransaction).transaction(async () => { + // Store mapping + await lidMapping.storeLIDPNMapping(toJid, fromJid) + + // Load and copy session + const fromAddr = jidToSignalProtocolAddress(fromJid) + const fromSession = await storage.loadSession(fromAddr.toString()) + + if (fromSession?.haveOpenSession()) { + // Deep copy session to prevent reference issues + const sessionBytes = fromSession.serialize() + const copiedSession = libsignal.SessionRecord.deserialize(sessionBytes) + + // Store at LID address + await storage.storeSession(lidAddr.toString(), copiedSession) + + // Delete PN session - maintain single encryption layer + await auth.keys.set({ session: { [fromAddr.toString()]: null } }) + } + + recentMigrations.set(migrationKey, true) + }) + }, + + async encryptMessageWithWire({ encryptionJid, wireJid, data }) { + const result = await repository.encryptMessage({ jid: encryptionJid, data }) + return { ...result, wireJid } + }, + + destroy() { + recentMigrations.clear() } } + + return repository } const jidToSignalProtocolAddress = (jid: string) => { - const { user, device } = jidDecode(jid)! - return new libsignal.ProtocolAddress(user, device || 0) + const decoded = jidDecode(jid)! + const { user, device, server } = decoded + + // LID addresses get _1 suffix for Signal protocol + const signalUser = server === 'lid' ? `${user}_1` : user + const finalDevice = device || 0 + + return new libsignal.ProtocolAddress(signalUser, finalDevice) } const jidToSignalSenderKeyName = (group: string, user: string): SenderKeyName => { return new SenderKeyName(group, jidToSignalProtocolAddress(user)) } -function signalStorage({ creds, keys }: SignalAuthState): SenderKeyStore & Record { +function signalStorage( + { creds, keys }: SignalAuthState, + lidMapping: LIDMappingStore +): SenderKeyStore & Record { return { loadSession: async (id: string) => { - const { [id]: sess } = await keys.get('session', [id]) - if (sess) { - return libsignal.SessionRecord.deserialize(sess) + try { + // LID SINGLE SOURCE OF TRUTH: Auto-redirect PN to LID if mapping exists + let actualId = id + if (id.includes('.') && !id.includes('_1')) { + // This is a PN signal address format (e.g., "1234567890.0") + // Convert back to JID to check for LID mapping + const parts = id.split('.') + const device = parts[1] || '0' + const pnJid = device === '0' ? `${parts[0]}@s.whatsapp.net` : `${parts[0]}:${device}@s.whatsapp.net` + + const lidForPN = await lidMapping.getLIDForPN(pnJid) + if (lidForPN?.includes('@lid')) { + const lidAddr = jidToSignalProtocolAddress(lidForPN) + const lidId = lidAddr.toString() + + // Check if LID session exists + const { [lidId]: lidSession } = await keys.get('session', [lidId]) + if (lidSession) { + actualId = lidId + } + } + } + + const { [actualId]: sess } = await keys.get('session', [actualId]) + + if (sess) { + return libsignal.SessionRecord.deserialize(sess) + } + } catch (e) { + return null } + + return null }, // TODO: Replace with libsignal.SessionRecord when type exports are added to libsignal storeSession: async (id: string, session: any) => { diff --git a/src/Signal/lid-mapping.ts b/src/Signal/lid-mapping.ts new file mode 100644 index 00000000..e902f0d4 --- /dev/null +++ b/src/Signal/lid-mapping.ts @@ -0,0 +1,101 @@ +import type { SignalKeyStoreWithTransaction } from '../Types' +import logger from '../Utils/logger' +import { isJidUser, isLidUser, jidDecode } from '../WABinary' + +export class LIDMappingStore { + private readonly keys: SignalKeyStoreWithTransaction + + constructor(keys: SignalKeyStoreWithTransaction) { + this.keys = keys + } + + /** + * Store LID-PN mapping - USER LEVEL + */ + async storeLIDPNMapping(lid: string, pn: string): Promise { + // Validate inputs + if (!((isLidUser(lid) && isJidUser(pn)) || (isJidUser(lid) && isLidUser(pn)))) { + logger.warn(`Invalid LID-PN mapping: ${lid}, ${pn}`) + return + } + + const [lidJid, pnJid] = isLidUser(lid) ? [lid, pn] : [pn, lid] + + const lidDecoded = jidDecode(lidJid) + const pnDecoded = jidDecode(pnJid) + + if (!lidDecoded || !pnDecoded) return + + const pnUser = pnDecoded.user + const lidUser = lidDecoded.user + + logger.trace(`Storing USER LID mapping: PN ${pnUser} → LID ${lidUser}`) + + await this.keys.transaction(async () => { + await this.keys.set({ + 'lid-mapping': { + [pnUser]: lidUser, // "554396160286" -> "102765716062358" + [`${lidUser}_reverse`]: pnUser // "102765716062358_reverse" -> "554396160286" + } + }) + }) + + logger.trace(`USER LID mapping stored: PN ${pnUser} → LID ${lidUser}`) + } + + /** + * Get LID for PN - Returns device-specific LID based on user mapping + */ + async getLIDForPN(pn: string): Promise { + if (!isJidUser(pn)) return null + + const decoded = jidDecode(pn) + if (!decoded) return null + + // Look up user-level mapping (whatsmeow approach) + const pnUser = decoded.user + const stored = await this.keys.get('lid-mapping', [pnUser]) + const lidUser = stored[pnUser] + + if (!lidUser) { + logger.trace(`No LID mapping found for PN user ${pnUser}`) + return null + } + + if (typeof lidUser !== 'string') return null + + // Push the PN device ID to the LID to maintain device separation + const pnDevice = decoded.device !== undefined ? decoded.device : 0 + const deviceSpecificLid = `${lidUser}:${pnDevice}@lid` + + logger.trace(`getLIDForPN: ${pn} → ${deviceSpecificLid} (user mapping with device ${pnDevice})`) + return deviceSpecificLid + } + + /** + * Get PN for LID - USER LEVEL with device construction + */ + async getPNForLID(lid: string): Promise { + if (!isLidUser(lid)) return null + + const decoded = jidDecode(lid) + if (!decoded) return null + + // Look up reverse user mapping + const lidUser = decoded.user + const stored = await this.keys.get('lid-mapping', [`${lidUser}_reverse`]) + const pnUser = stored[`${lidUser}_reverse`] + + if (!pnUser || typeof pnUser !== 'string') { + logger.trace(`No reverse mapping found for LID user: ${lidUser}`) + return null + } + + // Construct device-specific PN JID + const lidDevice = decoded.device !== undefined ? decoded.device : 0 + const pnJid = `${pnUser}:${lidDevice}@s.whatsapp.net` + + logger.trace(`Found reverse mapping: ${lid} → ${pnJid}`) + return pnJid + } +} diff --git a/src/Socket/messages-recv.ts b/src/Socket/messages-recv.ts index c39fb870..0a2a894b 100644 --- a/src/Socket/messages-recv.ts +++ b/src/Socket/messages-recv.ts @@ -816,6 +816,45 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { ev.emit('chats.phoneNumberShare', { lid: node.attrs.from!, jid: node.attrs.sender_pn }) } + if (msg.message?.protocolMessage?.lidMigrationMappingSyncMessage?.encodedMappingPayload) { + try { + const payload = msg.message.protocolMessage.lidMigrationMappingSyncMessage.encodedMappingPayload + const decoded = proto.LIDMigrationMappingSyncPayload.decode(payload) + + logger.debug( + { + mappingCount: decoded.pnToLidMappings?.length || 0, + timestamp: decoded.chatDbMigrationTimestamp + }, + 'Received LID migration sync message from server' + ) + + const lidMapping = signalRepository.getLIDMappingStore() + if (decoded.pnToLidMappings && decoded.pnToLidMappings.length > 0) { + for (const mapping of decoded.pnToLidMappings) { + const pn = `${mapping.pn}@s.whatsapp.net` + // Use latestLid if available, otherwise assignedLid (proper LID refresh) + const lidValue = mapping.latestLid || mapping.assignedLid + const lid = `${lidValue}@lid` + + await lidMapping.storeLIDPNMapping(lid, pn) + logger.debug( + { + pn, + lid, + assignedLid: mapping.assignedLid, + latestLid: mapping.latestLid, + usedLatest: !!mapping.latestLid + }, + 'Stored server-provided PN-LID mapping' + ) + } + } + } catch (error) { + logger.error({ error }, 'Failed to process LID migration sync message') + } + } + try { await Promise.all([ processingMutex.mutex(async () => { diff --git a/src/Socket/messages-send.ts b/src/Socket/messages-send.ts index 4050d829..5e8e7e1b 100644 --- a/src/Socket/messages-send.ts +++ b/src/Socket/messages-send.ts @@ -31,6 +31,7 @@ import { unixTimestampSeconds } from '../Utils' import { getUrlInfo } from '../Utils/link-preview' +import { makeKeyedMutex } from '../Utils/make-mutex' import { areJidsSameUser, type BinaryNode, @@ -80,6 +81,9 @@ export const makeMessagesSocket = (config: SocketConfig) => { useClones: false }) + // Prevent race conditions in Signal session encryption by user + const encryptionMutex = makeKeyedMutex() + let mediaConn: Promise const refreshMediaConn = async (forceGet = false) => { const media = await mediaConn @@ -185,24 +189,87 @@ export const makeMessagesSocket = (config: SocketConfig) => { await sendReceipts(keys, readType) } + /** Device info with wire JID format for envelope addressing */ + type DeviceWithWireJid = JidWithDevice & { + wireJid: string // The exact JID format that should be used in wire protocol (envelope addressing) + } + + /** + * Deduplicate JIDs when both LID and PN versions exist for same user + * Prefers LID over PN to maintain single encryption layer + */ + const deduplicateLidPnJids = (jids: string[]): string[] => { + const lidUsers = new Set() + const filteredJids: string[] = [] + + // Collect all LID users + for (const jid of jids) { + if (jid.includes('@lid')) { + const user = jidDecode(jid)?.user + if (user) lidUsers.add(user) + } + } + + // Filter out PN versions when LID exists + for (const jid of jids) { + if (jid.includes('@s.whatsapp.net')) { + const user = jidDecode(jid)?.user + if (user && lidUsers.has(user)) { + logger.debug({ jid }, 'Skipping PN - LID version exists') + continue + } + } + + filteredJids.push(jid) + } + + return filteredJids + } + /** Fetch all the devices we've to send a message to */ - const getUSyncDevices = async (jids: string[], useCache: boolean, ignoreZeroDevices: boolean) => { - const deviceResults: JidWithDevice[] = [] + const getUSyncDevices = async ( + jids: string[], + useCache: boolean, + ignoreZeroDevices: boolean + ): Promise => { + const deviceResults: DeviceWithWireJid[] = [] if (!useCache) { logger.debug('not using cache for devices') } const toFetch: string[] = [] - jids = Array.from(new Set(jids)) + // Deduplicate and normalize JIDs + jids = deduplicateLidPnJids(Array.from(new Set(jids))) for (let jid of jids) { - const user = jidDecode(jid)?.user + const decoded = jidDecode(jid) + const user = decoded?.user + const device = decoded?.device + const isExplicitDevice = typeof device === 'number' && device >= 0 + + // Handle explicit device JIDs directly + if (isExplicitDevice && user) { + deviceResults.push({ + user, + device, + wireJid: jid // Preserve exact JID format for wire protocol + }) + continue + } + + // For user JIDs, normalize and prepare for device enumeration jid = jidNormalizedUser(jid) + if (useCache) { - const devices = userDevicesCache.get(user!) + const devices = userDevicesCache.get(user!) as JidWithDevice[] if (devices) { - deviceResults.push(...devices) + const isLidJid = jid.includes('@lid') + const devicesWithWire = devices.map(d => ({ + ...d, + wireJid: isLidJid ? jidEncode(d.user, 'lid', d.device) : jidEncode(d.user, 's.whatsapp.net', d.device) + })) + deviceResults.push(...devicesWithWire) logger.trace({ user }, 'using cache for devices') } else { @@ -217,6 +284,14 @@ export const makeMessagesSocket = (config: SocketConfig) => { return deviceResults } + const requestedLidUsers = new Set() + for (const jid of toFetch) { + if (jid.includes('@lid')) { + const user = jidDecode(jid)?.user + if (user) requestedLidUsers.add(user) + } + } + const query = new USyncQuery().withContext('message').withDeviceProtocol() for (const jid of toFetch) { @@ -232,8 +307,33 @@ export const makeMessagesSocket = (config: SocketConfig) => { for (const item of extracted) { deviceMap[item.user] = deviceMap[item.user] || [] deviceMap[item.user]?.push(item) + } - deviceResults.push(item) + // Process each user's devices as a group for bulk LID migration + for (const [user, userDevices] of Object.entries(deviceMap)) { + const isLidUser = requestedLidUsers.has(user) + + // Process all devices for this user + for (const item of userDevices) { + const finalWireJid = isLidUser + ? jidEncode(user, 'lid', item.device) + : jidEncode(item.user, 's.whatsapp.net', item.device) + + deviceResults.push({ + ...item, + wireJid: finalWireJid + }) + + logger.debug( + { + user: item.user, + device: item.device, + finalWireJid, + usedLid: isLidUser + }, + 'Processed device with LID priority' + ) + } } for (const key in deviceMap) { @@ -244,24 +344,216 @@ export const makeMessagesSocket = (config: SocketConfig) => { return deviceResults } + // Helper to check if JID has migrated LID session + const checkForMigratedLidSession = async (jid: string): Promise => { + if (!jid.includes('@s.whatsapp.net')) return false + + const lidMapping = signalRepository.getLIDMappingStore() + const lidForPN = await lidMapping.getLIDForPN(jid) + if (!lidForPN?.includes('@lid')) return false + + const lidSignalId = signalRepository.jidToSignalProtocolAddress(lidForPN) + const lidSessions = await authState.keys.get('session', [lidSignalId]) + return !!lidSessions[lidSignalId] + } + const assertSessions = async (jids: string[], force: boolean) => { let didFetchNewSession = false - let jidsRequiringFetch: string[] = [] + const jidsRequiringFetch: string[] = [] + + // Apply same deduplication as in getUSyncDevices + jids = deduplicateLidPnJids(jids) + if (force) { - jidsRequiringFetch = jids - } else { + // Check which sessions are missing (with LID migration check) const addrs = jids.map(jid => signalRepository.jidToSignalProtocolAddress(jid)) const sessions = await authState.keys.get('session', addrs) - for (const jid of jids) { + + // Helper to check session for a JID + const checkJidSession = async (jid: string) => { const signalId = signalRepository.jidToSignalProtocolAddress(jid) - if (!sessions[signalId]) { + let hasSession = !!sessions[signalId] + + // Check for migrated LID session if PN session missing + if (!hasSession) { + hasSession = await checkForMigratedLidSession(jid) + if (hasSession) { + logger.debug({ jid }, 'Found migrated LID session during force assert, skipping PN fetch') + } + } + + // Add to fetch list if no session exists + if (!hasSession) { + if (jid.includes('@lid')) { + logger.debug({ jid }, 'No LID session found, will create new LID session') + } + jidsRequiringFetch.push(jid) } } + + // Process all JIDs + for (const jid of jids) { + await checkJidSession(jid) + } + } else { + const lidMapping = signalRepository.getLIDMappingStore() + const addrs = jids.map(jid => signalRepository.jidToSignalProtocolAddress(jid)) + const sessions = await authState.keys.get('session', addrs) + + // Group JIDs by user for bulk migration + const userGroups = new Map() + for (const jid of jids) { + const user = jidNormalizedUser(jid) + if (!userGroups.has(user)) { + userGroups.set(user, []) + } + + userGroups.get(user)!.push(jid) + } + + // Helper to check LID mapping for a user + const checkUserLidMapping = async (user: string, userJids: string[]) => { + if (!userJids.some(jid => jid.includes('@s.whatsapp.net'))) { + return { shouldMigrate: false, lidForPN: undefined } + } + + try { + const mapping = await lidMapping.getLIDForPN(user) + if (mapping?.includes('@lid')) { + logger.debug( + { user, lidForPN: mapping, deviceCount: userJids.length }, + 'User has LID mapping - preparing bulk migration' + ) + return { shouldMigrate: true, lidForPN: mapping } + } + } catch (error) { + logger.debug({ user, error }, 'Failed to check LID mapping for user') + } + + return { shouldMigrate: false, lidForPN: undefined } + } + + // Helper to migrate a single device + const migrateDeviceToLid = async (jid: string, lidForPN: string) => { + if (!jid.includes('@s.whatsapp.net')) return + + try { + const jidDecoded = jidDecode(jid) + const deviceId = jidDecoded?.device || 0 + const lidDecoded = jidDecode(lidForPN) + const lidWithDevice = jidEncode(lidDecoded?.user!, 'lid', deviceId) + + await signalRepository.migrateSession(jid, lidWithDevice) + logger.debug({ fromJid: jid, toJid: lidWithDevice }, 'Migrated device session to LID') + + // Delete PN session after successful migration + try { + await signalRepository.deleteSession(jid) + logger.debug({ deletedPNSession: jid }, 'Deleted PN session after migration') + } catch (deleteError) { + logger.warn({ jid, error: deleteError }, 'Failed to delete PN session') + } + } catch (migrationError) { + logger.warn({ jid, error: migrationError }, 'Failed to migrate device session') + } + } + + // Process each user group for potential bulk LID migration + for (const [user, userJids] of userGroups) { + const mappingResult = await checkUserLidMapping(user, userJids) + const shouldMigrateUser = mappingResult.shouldMigrate + const lidForPN = mappingResult.lidForPN + + // Migrate all devices for this user if LID mapping exists + if (shouldMigrateUser && lidForPN) { + // Migrate each device individually + for (const jid of userJids) { + await migrateDeviceToLid(jid, lidForPN) + } + + logger.info( + { + user, + lidMapping: lidForPN, + deviceCount: userJids.length + }, + 'Completed migration attempt for user devices' + ) + } + + // Helper to check session for migrated user + const checkMigratedSession = async (jid: string) => { + const signalId = signalRepository.jidToSignalProtocolAddress(jid) + let hasSession = !!sessions[signalId] + let jidToFetch = jid + + // Check if we should use migrated LID session instead + if (shouldMigrateUser && lidForPN && jid.includes('@s.whatsapp.net')) { + const originalDecoded = jidDecode(jid) + const deviceId = originalDecoded?.device || 0 + const lidDecoded = jidDecode(lidForPN) + const lidWithDevice = jidEncode(lidDecoded?.user!, 'lid', deviceId) + + // Check if LID session exists + const lidSignalId = signalRepository.jidToSignalProtocolAddress(lidWithDevice) + const lidSessions = await authState.keys.get('session', [lidSignalId]) + hasSession = !!lidSessions[lidSignalId] + jidToFetch = lidWithDevice + + if (hasSession) { + logger.debug({ originalJid: jid, lidJid: lidWithDevice }, '✅ Found bulk-migrated LID session') + } + } + + // Add to fetch list if no session exists + if (!hasSession) { + jidsRequiringFetch.push(jidToFetch) + logger.debug( + { jid: jidToFetch, originalJid: jid !== jidToFetch ? jid : undefined }, + 'Adding to session fetch list' + ) + } + } + + // Now check which sessions need to be fetched for this user + for (const jid of userJids) { + await checkMigratedSession(jid) + } + } } if (jidsRequiringFetch.length) { logger.debug({ jidsRequiringFetch }, 'fetching sessions') + + // DEBUG: Check if there are PN versions of LID users being fetched + const lidUsersBeingFetched = new Set() + const pnUsersBeingFetched = new Set() + + for (const jid of jidsRequiringFetch) { + const user = jidDecode(jid)?.user + if (user) { + if (jid.includes('@lid')) { + lidUsersBeingFetched.add(user) + } else if (jid.includes('@s.whatsapp.net')) { + pnUsersBeingFetched.add(user) + } + } + } + + // Find overlaps + const overlapping = Array.from(pnUsersBeingFetched).filter(user => lidUsersBeingFetched.has(user)) + if (overlapping.length > 0) { + logger.warn( + { + overlapping, + lidUsersBeingFetched: Array.from(lidUsersBeingFetched), + pnUsersBeingFetched: Array.from(pnUsersBeingFetched) + }, + 'Fetching both LID and PN sessions for same users' + ) + } + const result = await query({ tag: 'iq', attrs: { @@ -316,7 +608,12 @@ export const makeMessagesSocket = (config: SocketConfig) => { return msgId } - const createParticipantNodes = async (jids: string[], message: proto.IMessage, extraAttrs?: BinaryNode['attrs']) => { + const createParticipantNodes = async ( + jids: string[], + message: proto.IMessage, + extraAttrs?: BinaryNode['attrs'], + dsmMessage?: proto.IMessage + ) => { let patched = await patchMessageBeforeSending(message, jids) if (!Array.isArray(patched)) { patched = jids ? jids.map(jid => ({ recipientJid: jid, ...patched })) : [patched] @@ -324,37 +621,142 @@ export const makeMessagesSocket = (config: SocketConfig) => { let shouldIncludeDeviceIdentity = false - const nodes = await Promise.all( - patched.map(async patchedMessageWithJid => { - const { recipientJid: jid, ...patchedMessage } = patchedMessageWithJid - if (!jid) { - return {} as BinaryNode - } + const meId = authState.creds.me!.id + const meLid = authState.creds.me?.lid + const meLidUser = meLid ? jidDecode(meLid)?.user : null - const bytes = encodeWAMessage(patchedMessage) - const { type, ciphertext } = await signalRepository.encryptMessage({ jid, data: bytes }) - if (type === 'pkmsg') { - shouldIncludeDeviceIdentity = true - } + const devicesByUser = new Map>() - const node: BinaryNode = { - tag: 'to', - attrs: { jid }, - content: [ - { - tag: 'enc', - attrs: { - v: '2', - type, - ...(extraAttrs || {}) - }, - content: ciphertext + for (const patchedMessageWithJid of patched) { + const { recipientJid: wireJid, ...patchedMessage } = patchedMessageWithJid + if (!wireJid) continue + + // Extract user from JID for grouping + const decoded = jidDecode(wireJid) + const user = decoded?.user + + if (!user) continue + + if (!devicesByUser.has(user)) { + devicesByUser.set(user, []) + } + + devicesByUser.get(user)!.push({ recipientJid: wireJid, patchedMessage }) + } + + // Process each user's devices sequentially, but different users in parallel + const userEncryptionPromises = Array.from(devicesByUser.entries()).map(([user, userDevices]) => + encryptionMutex.mutex(user, async () => { + logger.debug({ user, deviceCount: userDevices.length }, 'Acquiring encryption lock for user devices') + + const userNodes: BinaryNode[] = [] + + // Helper to get encryption JID with LID migration + const getEncryptionJid = async (wireJid: string) => { + if (!wireJid.includes('@s.whatsapp.net')) return wireJid + + try { + const lidMapping = signalRepository.getLIDMappingStore() + const lidForPN = await lidMapping.getLIDForPN(wireJid) + + if (!lidForPN?.includes('@lid')) return wireJid + + // Preserve device ID from original wire JID + const wireDecoded = jidDecode(wireJid) + const deviceId = wireDecoded?.device || 0 + const lidDecoded = jidDecode(lidForPN) + const lidWithDevice = jidEncode(lidDecoded?.user!, 'lid', deviceId) + + // Migrate session to LID for unified encryption layer + try { + await signalRepository.migrateSession(wireJid, lidWithDevice) + const recipientUser = jidNormalizedUser(wireJid) + const ownPnUser = jidNormalizedUser(meId) + const isOwnDevice = recipientUser === ownPnUser + logger.info({ wireJid, lidWithDevice, isOwnDevice }, 'Migrated to LID encryption') + + // Delete PN session after successful migration + try { + await signalRepository.deleteSession(wireJid) + logger.debug({ deletedPNSession: wireJid }, 'Deleted PN session') + } catch (deleteError) { + logger.warn({ wireJid, error: deleteError }, 'Failed to delete PN session') + } + + return lidWithDevice + } catch (migrationError) { + logger.warn({ wireJid, error: migrationError }, 'Failed to migrate session') + return wireJid } - ] + } catch (error) { + logger.debug({ wireJid, error }, 'Failed to check LID mapping') + return wireJid + } } - return node + + // Encrypt to this user's devices sequentially to prevent session corruption + for (const { recipientJid: wireJid, patchedMessage } of userDevices) { + // DSM logic: Use DSM for own other devices (following whatsmeow implementation) + let messageToEncrypt = patchedMessage + if (dsmMessage) { + const { user: targetUser } = jidDecode(wireJid)! + const { user: ownPnUser } = jidDecode(meId)! + const ownLidUser = meLidUser + + // Check if this is our device (same user, different device) + const isOwnUser = targetUser === ownPnUser || (ownLidUser && targetUser === ownLidUser) + + // Exclude exact sender device (whatsmeow: if jid == ownJID || jid == ownLID { continue }) + const isExactSenderDevice = + wireJid === meId || (authState.creds.me?.lid && wireJid === authState.creds.me.lid) + + if (isOwnUser && !isExactSenderDevice) { + messageToEncrypt = dsmMessage + logger.debug({ wireJid, targetUser }, 'Using DSM for own device') + } + } + + const bytes = encodeWAMessage(messageToEncrypt) + + // Get encryption JID with LID migration + const encryptionJid = await getEncryptionJid(wireJid) + + // ENCRYPT: Use the determined encryption identity (prefers migrated LID) + const { type, ciphertext } = await signalRepository.encryptMessage({ + jid: encryptionJid, // Unified encryption layer (LID when available) + data: bytes + }) + + if (type === 'pkmsg') { + shouldIncludeDeviceIdentity = true + } + + const node: BinaryNode = { + tag: 'to', + attrs: { jid: wireJid }, // Always use original wire identity in envelope + content: [ + { + tag: 'enc', + attrs: { + v: '2', + type, + ...(extraAttrs || {}) + }, + content: ciphertext + } + ] + } + userNodes.push(node) + } + + logger.debug({ user, nodesCreated: userNodes.length }, 'Releasing encryption lock for user devices') + return userNodes }) ) + + // Wait for all users to complete (users are processed in parallel) + const userNodesArrays = await Promise.all(userEncryptionPromises) + const nodes = userNodesArrays.flat() return { nodes, shouldIncludeDeviceIdentity } } @@ -372,6 +774,9 @@ export const makeMessagesSocket = (config: SocketConfig) => { }: MessageRelayOptions ) => { const meId = authState.creds.me!.id + const meLid = authState.creds.me?.lid + + // ADDRESSING CONSISTENCY: Keep envelope addressing as user provided, handle LID migration in encryption let shouldIncludeDeviceIdentity = false @@ -382,14 +787,27 @@ export const makeMessagesSocket = (config: SocketConfig) => { const isLid = server === 'lid' const isNewsletter = server === 'newsletter' + // Keep user's original JID choice for envelope addressing + const finalJid = jid + + // ADDRESSING CONSISTENCY: Match own identity to conversation context + let ownId = meId + if (isLid && meLid) { + ownId = meLid + logger.debug({ to: jid, ownId }, 'Using LID identity for @lid conversation') + } else { + logger.debug({ to: jid, ownId }, 'Using PN identity for @s.whatsapp.net conversation') + } + msgId = msgId || generateMessageIDV2(sock.user?.id) useUserDevicesCache = useUserDevicesCache !== false useCachedGroupMetadata = useCachedGroupMetadata !== false && !isStatus const participants: BinaryNode[] = [] - const destinationJid = !isStatus ? jidEncode(user, isLid ? 'lid' : isGroup ? 'g.us' : 's.whatsapp.net') : statusJid + const destinationJid = !isStatus ? finalJid : statusJid + const binaryNodeContent: BinaryNode[] = [] - const devices: JidWithDevice[] = [] + const devices: DeviceWithWireJid[] = [] const meMsg: proto.IMessage = { deviceSentMessage: { @@ -410,7 +828,11 @@ export const makeMessagesSocket = (config: SocketConfig) => { } const { user, device } = jidDecode(participant.jid)! - devices.push({ user, device }) + devices.push({ + user, + device, + wireJid: participant.jid // Use the participant JID as wire JID + }) } await authState.keys.transaction(async () => { @@ -476,9 +898,10 @@ export const makeMessagesSocket = (config: SocketConfig) => { } if (!isStatus) { + const groupAddressingMode = groupData?.addressingMode || (isLid ? 'lid' : 'pn') additionalAttributes = { ...additionalAttributes, - addressing_mode: groupData?.addressingMode || 'pn' + addressing_mode: groupAddressingMode } } @@ -494,20 +917,26 @@ export const makeMessagesSocket = (config: SocketConfig) => { const bytes = encodeWAMessage(patched) + // This should match the group's addressing mode and conversation context + const groupAddressingMode = groupData?.addressingMode || (isLid ? 'lid' : 'pn') + const groupSenderIdentity = groupAddressingMode === 'lid' && meLid ? meLid : meId + const { ciphertext, senderKeyDistributionMessage } = await signalRepository.encryptGroupMessage({ group: destinationJid, data: bytes, - meId + meId: groupSenderIdentity }) const senderKeyJids: string[] = [] // ensure a connection is established with every device - for (const { user, device } of devices) { - const jid = jidEncode(user, groupData?.addressingMode === 'lid' ? 'lid' : 's.whatsapp.net', device) - if (!senderKeyMap[jid] || !!participant) { - senderKeyJids.push(jid) + for (const device of devices) { + // This preserves the LID migration results from getUSyncDevices + const deviceJid = device.wireJid + const hasKey = !!senderKeyMap[deviceJid] + if (!hasKey || !!participant) { + senderKeyJids.push(deviceJid) // store that this person has had the sender keys sent to them - senderKeyMap[jid] = true + senderKeyMap[deviceJid] = true } } @@ -539,30 +968,70 @@ export const makeMessagesSocket = (config: SocketConfig) => { await authState.keys.set({ 'sender-key-memory': { [jid]: senderKeyMap } }) } else { - const { user: meUser } = jidDecode(meId)! + const { user: ownUser } = jidDecode(ownId)! if (!participant) { - devices.push({ user }) - if (user !== meUser) { - devices.push({ user: meUser }) + const targetUserServer = isLid ? 'lid' : 's.whatsapp.net' + devices.push({ + user, + device: 0, + wireJid: jidEncode(user, targetUserServer, 0) + }) + + // Own user matches conversation addressing mode + if (user !== ownUser) { + const ownUserServer = isLid ? 'lid' : 's.whatsapp.net' + const ownUserForAddressing = isLid && meLid ? jidDecode(meLid)!.user : jidDecode(meId)!.user + + devices.push({ + user: ownUserForAddressing, + device: 0, + wireJid: jidEncode(ownUserForAddressing, ownUserServer, 0) + }) } if (additionalAttributes?.['category'] !== 'peer') { - const additionalDevices = await getUSyncDevices([meId, jid], !!useUserDevicesCache, true) - devices.push(...additionalDevices) + // Clear placeholders and enumerate actual devices + devices.length = 0 + + // Use conversation-appropriate sender identity + const senderIdentity = + isLid && meLid + ? jidEncode(jidDecode(meLid)?.user!, 'lid', undefined) + : jidEncode(jidDecode(meId)?.user!, 's.whatsapp.net', undefined) + + // Enumerate devices for sender and target with consistent addressing + const sessionDevices = await getUSyncDevices([senderIdentity, jid], false, false) + devices.push(...sessionDevices) + + logger.debug( + { + deviceCount: devices.length, + devices: devices.map(d => `${d.user}:${d.device}@${jidDecode(d.wireJid)?.server}`) + }, + 'Device enumeration complete with unified addressing' + ) } } const allJids: string[] = [] const meJids: string[] = [] const otherJids: string[] = [] - for (const { user, device } of devices) { - const isMe = user === meUser - const jid = jidEncode( - isMe && isLid ? authState.creds?.me?.lid!.split(':')[0] || user : user, - isLid ? 'lid' : 's.whatsapp.net', - device - ) + const { user: mePnUser } = jidDecode(meId)! + const { user: meLidUser } = meLid ? jidDecode(meLid)! : { user: null } + + for (const { user, wireJid } of devices) { + const isExactSenderDevice = wireJid === meId || (meLid && wireJid === meLid) + if (isExactSenderDevice) { + logger.debug({ wireJid, meId, meLid }, 'Skipping exact sender device (whatsmeow pattern)') + continue + } + + // Check if this is our device (could match either PN or LID user) + const isMe = user === mePnUser || (meLidUser && user === meLidUser) + + const jid = wireJid + if (isMe) { meJids.push(jid) } else { @@ -572,14 +1041,15 @@ export const makeMessagesSocket = (config: SocketConfig) => { allJids.push(jid) } - await assertSessions(allJids, false) + await assertSessions([...otherJids, ...meJids], false) const [ { nodes: meNodes, shouldIncludeDeviceIdentity: s1 }, { nodes: otherNodes, shouldIncludeDeviceIdentity: s2 } ] = await Promise.all([ - createParticipantNodes(meJids, meMsg, extraAttrs), - createParticipantNodes(otherJids, message, extraAttrs) + // For own devices: use DSM if available (1:1 chats only) + createParticipantNodes(meJids, meMsg || message, extraAttrs), + createParticipantNodes(otherJids, message, extraAttrs, meMsg) ]) participants.push(...meNodes) participants.push(...otherNodes) @@ -597,6 +1067,7 @@ export const makeMessagesSocket = (config: SocketConfig) => { binaryNodeContent.push({ tag: 'participants', attrs: {}, + content: participants }) } @@ -606,11 +1077,13 @@ export const makeMessagesSocket = (config: SocketConfig) => { tag: 'message', attrs: { id: msgId, + to: destinationJid, type: getMessageType(message), ...(additionalAttributes || {}) }, content: binaryNodeContent } + // if the participant to send to is explicitly specified (generally retry recp) // ensure the message is only sent to that person // if a retry receipt is sent to everyone -- it'll fail decryption for everyone else who received the msg diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index 4c238a66..76263b50 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -759,6 +759,25 @@ export const makeSocket = (config: SocketConfig) => { ev.emit('creds.update', { me: { ...authState.creds.me!, lid: node.attrs.lid } }) ev.emit('connection.update', { connection: 'open' }) + + if (node.attrs.lid && authState.creds.me?.id) { + const myLID = node.attrs.lid + process.nextTick(async () => { + try { + const myPN = authState.creds.me!.id + + // Store our own LID-PN mapping + await signalRepository.storeLIDPNMapping(myLID, myPN) + + // Create LID session for ourselves (whatsmeow pattern) + await signalRepository.migrateSession(myPN, myLID) + + logger.info({ myPN, myLID }, 'Own LID session created successfully') + } catch (error) { + logger.error({ error, lid: myLID }, 'Failed to create own LID session') + } + }) + } }) ws.on('CB:stream:error', (node: BinaryNode) => { diff --git a/src/Types/Auth.ts b/src/Types/Auth.ts index e4b5a366..187ba87b 100644 --- a/src/Types/Auth.ts +++ b/src/Types/Auth.ts @@ -72,6 +72,7 @@ export type SignalDataTypeMap = { 'sender-key-memory': { [jid: string]: boolean } 'app-state-sync-key': proto.Message.IAppStateSyncKeyData 'app-state-sync-version': LTHashState + 'lid-mapping': string } export type SignalDataSet = { [T in keyof SignalDataTypeMap]?: { [id: string]: SignalDataTypeMap[T] | null } } diff --git a/src/Types/Signal.ts b/src/Types/Signal.ts index f514f562..bd389ff4 100644 --- a/src/Types/Signal.ts +++ b/src/Types/Signal.ts @@ -1,4 +1,5 @@ import { proto } from '../../WAProto/index.js' +import type { LIDMappingStore } from '../Signal/lid-mapping' type DecryptGroupSignalOpts = { group: string @@ -22,6 +23,12 @@ type EncryptMessageOpts = { data: Uint8Array } +type EncryptMessageWithWireOpts = { + encryptionJid: string // JID used for session lookup (LID) + wireJid: string // JID used for envelope (PN) + data: Uint8Array +} + type EncryptGroupMessageOpts = { group: string data: Uint8Array @@ -57,10 +64,21 @@ export type SignalRepository = { type: 'pkmsg' | 'msg' ciphertext: Uint8Array }> + encryptMessageWithWire(opts: EncryptMessageWithWireOpts): Promise<{ + type: 'pkmsg' | 'msg' + ciphertext: Uint8Array + wireJid: string // Return the wire JID for envelope + }> encryptGroupMessage(opts: EncryptGroupMessageOpts): Promise<{ senderKeyDistributionMessage: Uint8Array ciphertext: Uint8Array }> injectE2ESession(opts: E2ESessionOpts): Promise jidToSignalProtocolAddress(jid: string): string + storeLIDPNMapping(lid: string, pn: string): Promise + getLIDMappingStore(): LIDMappingStore + migrateSession(fromJid: string, toJid: string): Promise + validateSession(jid: string): Promise<{ exists: boolean; reason?: string }> + deleteSession(jid: string): Promise + destroy(): void } diff --git a/src/Utils/decode-wa-message.ts b/src/Utils/decode-wa-message.ts index 394be1b9..90f61839 100644 --- a/src/Utils/decode-wa-message.ts +++ b/src/Utils/decode-wa-message.ts @@ -10,11 +10,51 @@ import { isJidNewsletter, isJidStatusBroadcast, isJidUser, - isLidUser + isLidUser, + jidDecode, + jidEncode, + jidNormalizedUser } from '../WABinary' import { unpadRandomMax16 } from './generics' import type { ILogger } from './logger' +const getDecryptionJid = async (sender: string, repository: SignalRepository): Promise => { + if (!sender.includes('@s.whatsapp.net')) { + return sender + } + + const lidMapping = repository.getLIDMappingStore() + const normalizedSender = jidNormalizedUser(sender) + const lidForPN = await lidMapping.getLIDForPN(normalizedSender) + + if (lidForPN?.includes('@lid')) { + const senderDecoded = jidDecode(sender) + const deviceId = senderDecoded?.device || 0 + return jidEncode(jidDecode(lidForPN)!.user, 'lid', deviceId) + } + + return sender +} + +const storeMappingFromEnvelope = async ( + stanza: BinaryNode, + sender: string, + decryptionJid: string, + repository: SignalRepository, + logger: ILogger +): Promise => { + const { senderAlt } = extractAddressingContext(stanza) + + if (senderAlt && isLidUser(senderAlt) && isJidUser(sender) && decryptionJid === sender) { + try { + await repository.storeLIDPNMapping(senderAlt, sender) + logger.debug({ sender, senderAlt }, 'Stored LID mapping from envelope') + } catch (error) { + logger.warn({ sender, senderAlt, error }, 'Failed to store LID mapping') + } + } +} + export const NO_MESSAGE_FOUND_ERROR_TEXT = 'Message absent from node' export const MISSING_KEYS_ERROR_TEXT = 'Key used already or never filled' @@ -43,6 +83,28 @@ type MessageType = | 'other_status' | 'newsletter' +export const extractAddressingContext = (stanza: BinaryNode) => { + const addressingMode = stanza.attrs.addressing_mode || 'pn' + let senderAlt: string | undefined + let recipientAlt: string | undefined + + if (addressingMode === 'lid') { + // Message is LID-addressed: sender is LID, extract corresponding PN + senderAlt = stanza.attrs.participant_pn || stanza.attrs.sender_pn + recipientAlt = stanza.attrs.recipient_pn + } else { + // Message is PN-addressed: sender is PN, extract corresponding LID + senderAlt = stanza.attrs.participant_lid || stanza.attrs.sender_lid + recipientAlt = stanza.attrs.recipient_lid + } + + return { + addressingMode, + senderAlt, + recipientAlt + } +} + /** * Decode the received node as a message. * @note this will only parse the message, not decrypt it @@ -187,11 +249,15 @@ export const decryptMessageNode = ( case 'pkmsg': case 'msg': const user = isJidUser(sender) ? sender : author + const decryptionJid = await getDecryptionJid(user, repository) + msgBuffer = await repository.decryptMessage({ - jid: user, + jid: decryptionJid, type: e2eType, ciphertext: content }) + + await storeMappingFromEnvelope(stanza, user, decryptionJid, repository, logger) break case 'plaintext': msgBuffer = content diff --git a/yarn.lock b/yarn.lock index 6933c998..68d7e568 100644 --- a/yarn.lock +++ b/yarn.lock @@ -2863,6 +2863,7 @@ __metadata: json: "npm:^11.0.0" libsignal: "git+https://github.com/whiskeysockets/libsignal-node" link-preview-js: "npm:^3.0.0" + lru-cache: "npm:^10.4.3" music-metadata: "npm:^11.7.0" open: "npm:^8.4.2" pino: "npm:^9.6" @@ -7307,7 +7308,7 @@ __metadata: languageName: node linkType: hard -"lru-cache@npm:^10.0.1, lru-cache@npm:^10.2.0": +"lru-cache@npm:^10.0.1, lru-cache@npm:^10.2.0, lru-cache@npm:^10.4.3": version: 10.4.3 resolution: "lru-cache@npm:10.4.3" checksum: 10c0/ebd04fbca961e6c1d6c0af3799adcc966a1babe798f685bb84e6599266599cd95d94630b10262f5424539bc4640107e8a33aa28585374abf561d30d16f4b39fb