ruicesar/InfiniteAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix: groups pn messages-send (#1829)

Browse Source 
* fix: centralize wireJid at encryption layer

* fix: wirejid mess

* fix: use signalrepository validate session in assertsessions

* fix: wirejid conversion before fetch session

* fix: lint
This commit is contained in:
Gustavo Quadri
2025-09-27 23:45:47 -03:00
committed by GitHub
parent 1eef97aa86
commit 6f1faf81da
1 changed files with 62 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Socket/messages-send.ts
+62 -98
View File
@@ -200,37 +200,9 @@ export const makeMessagesSocket = (config: SocketConfig) => {
await sendReceipts(keys, readType)
}
/** Device info with wire JID format for envelope addressing */
type DeviceWithWireJid = JidWithDevice & {
wireJid: string
}
const resolveSessionJids = async (jids: string[]): Promise<Map<string, string>> => {
const uniquePnJids = Array.from(new Set(jids.filter(isPnUser)))
if (!uniquePnJids.length) {
return new Map()
}
const lookups = await Promise.all(
uniquePnJids.map(async pnJid => {
try {
const resolved = await signalRepository.lidMapping.getLIDForPN(pnJid)
return resolved ? ([pnJid, resolved] as const) : [pnJid, pnJid]
} catch (error) {
logger.warn({ pnJid, error }, 'Failed to resolve LID mapping for PN JID')
return [pnJid, pnJid]
}
})
)
const sessionMap = new Map<string, string>()
for (const entry of lookups) {
if (entry) {
sessionMap.set(entry[0], entry[1])
}
}
return sessionMap
/** Device info with wire JID */
type DeviceWithJid = JidWithDevice & {
jid: string
}
/** Fetch all the devices we've to send a message to */
@@ -238,8 +210,8 @@ export const makeMessagesSocket = (config: SocketConfig) => {
jids: string[],
useCache: boolean,
ignoreZeroDevices: boolean
): Promise<DeviceWithWireJid[]> => {
const deviceResults: DeviceWithWireJid[] = []
): Promise<DeviceWithJid[]> => {
const deviceResults: DeviceWithJid[] = []
if (!useCache) {
logger.debug('not using cache for devices')
@@ -258,7 +230,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
deviceResults.push({
user,
device,
wireJid: jid // again this makes no sense
jid
})
return null
}
@@ -282,11 +254,11 @@ export const makeMessagesSocket = (config: SocketConfig) => {
(userDevicesCache.mget ? undefined : ((await userDevicesCache.get(user!)) as JidWithDevice[]))
if (devices) {
const isLidJid = jid.includes('@lid')
const devicesWithWire = devices.map(d => ({
const devicesWithJid = devices.map(d => ({
...d,
wireJid: isLidJid ? jidEncode(d.user, 'lid', d.device) : jidEncode(d.user, 's.whatsapp.net', d.device)
jid: isLidJid ? jidEncode(d.user, 'lid', d.device) : jidEncode(d.user, 's.whatsapp.net', d.device)
}))
deviceResults.push(...devicesWithWire)
deviceResults.push(...devicesWithJid)
logger.trace({ user }, 'using cache for devices')
} else {
@@ -332,20 +304,20 @@ export const makeMessagesSocket = (config: SocketConfig) => {
// Process all devices for this user
for (const item of userDevices) {
const finalWireJid = isLidUser
const finalJid = isLidUser
? jidEncode(user, 'lid', item.device)
: jidEncode(item.user, 's.whatsapp.net', item.device)
deviceResults.push({
...item,
wireJid: finalWireJid
jid: finalJid
})
logger.debug(
{
user: item.user,
device: item.device,
finalWireJid,
finalJid,
usedLid: isLidUser
},
'Processed device with LID priority'
@@ -390,7 +362,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
const uniqueJids = [...new Set(jids)] // Deduplicate JIDs
const jidsRequiringFetch: string[] = []
// Check peerSessionsCache and authState.keys
// Check peerSessionsCache and validate sessions using libsignal loadSession
for (const jid of uniqueJids) {
const signalId = signalRepository.jidToSignalProtocolAddress(jid)
const cachedSession = peerSessionsCache.get(signalId)
@@ -399,8 +371,8 @@ export const makeMessagesSocket = (config: SocketConfig) => {
continue // Session exists in cache
}
} else {
const sessions = await authState.keys.get('session', [signalId])
const hasSession = !!sessions[signalId]
const sessionValidation = await signalRepository.validateSession(jid)
const hasSession = sessionValidation.exists
peerSessionsCache.set(signalId, hasSession)
if (hasSession) {
continue
@@ -411,7 +383,19 @@ export const makeMessagesSocket = (config: SocketConfig) => {
}
if (jidsRequiringFetch.length) {
logger.debug({ jidsRequiringFetch }, 'fetching sessions')
// LID if mapped, otherwise original
const wireJids = await Promise.all(
jidsRequiringFetch.map(async jid => {
if (jid.includes('@s.whatsapp.net')) {
const lid = await signalRepository.lidMapping.getLIDForPN(jid)
return lid ? lid : jid
}
return jid
})
)
logger.debug({ jidsRequiringFetch, wireJids }, 'fetching sessions')
const result = await query({
tag: 'iq',
attrs: {
@@ -423,7 +407,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
{
tag: 'key',
attrs: {},
content: jidsRequiringFetch.map(jid => ({
content: wireJids.map(jid => ({
tag: 'user',
attrs: { jid }
}))
@@ -433,9 +417,9 @@ export const makeMessagesSocket = (config: SocketConfig) => {
await parseAndInjectE2ESessions(result, signalRepository)
didFetchNewSession = true
// Cache fetched sessions
for (const jid of jidsRequiringFetch) {
const signalId = signalRepository.jidToSignalProtocolAddress(jid)
// Cache fetched sessions using wire JIDs
for (const wireJid of wireJids) {
const signalId = signalRepository.jidToSignalProtocolAddress(wireJid)
peerSessionsCache.set(signalId, true)
}
}
@@ -471,21 +455,20 @@ export const makeMessagesSocket = (config: SocketConfig) => {
return msgId
}
const createParticipantNodesWithSessionMap = async (
recipientWireJids: string[],
sessionMap: Map<string, string>,
const createParticipantNodes = async (
recipientJids: string[],
message: proto.IMessage,
extraAttrs?: BinaryNode['attrs'],
dsmMessage?: proto.IMessage
) => {
if (!recipientWireJids.length) {
if (!recipientJids.length) {
return { nodes: [] as BinaryNode[], shouldIncludeDeviceIdentity: false }
}
const patched = await patchMessageBeforeSending(message, recipientWireJids)
const patched = await patchMessageBeforeSending(message, recipientJids)
const patchedMessages = Array.isArray(patched)
? patched
: recipientWireJids.map(jid => ({ recipientJid: jid, message: patched }))
: recipientJids.map(jid => ({ recipientJid: jid, message: patched }))
let shouldIncludeDeviceIdentity = false
const meId = authState.creds.me!.id
@@ -493,27 +476,26 @@ export const makeMessagesSocket = (config: SocketConfig) => {
const meLidUser = meLid ? jidDecode(meLid)?.user : null
const encryptionPromises = (patchedMessages as any).map(
async ({ recipientJid: wireJid, message: patchedMessage }: any) => {
if (!wireJid) return null
wireJid = sessionMap.get(wireJid) ?? wireJid
async ({ recipientJid: jid, message: patchedMessage }: any) => {
if (!jid) return null
let msgToEncrypt = patchedMessage
if (dsmMessage) {
const { user: targetUser } = jidDecode(wireJid)!
const { user: targetUser } = jidDecode(jid)!
const { user: ownPnUser } = jidDecode(meId)!
const ownLidUser = meLidUser
const isOwnUser = targetUser === ownPnUser || (ownLidUser && targetUser === ownLidUser)
const isExactSenderDevice = wireJid === meId || (meLid && wireJid === meLid)
const isExactSenderDevice = jid === meId || (meLid && jid === meLid)
if (isOwnUser && !isExactSenderDevice) {
msgToEncrypt = dsmMessage
logger.debug({ wireJid, targetUser }, 'Using DSM for own device')
logger.debug({ jid, targetUser }, 'Using DSM for own device')
}
}
const bytes = encodeWAMessage(msgToEncrypt)
const mutexKey = wireJid
const mutexKey = jid
const node = await encryptionMutex.mutex(mutexKey, async () => {
const { type, ciphertext } = await signalRepository.encryptMessage({
jid: wireJid,
jid,
data: bytes
})
if (type === 'pkmsg') {
@@ -522,7 +504,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
return {
tag: 'to',
attrs: { jid: wireJid },
attrs: { jid },
content: [
{
tag: 'enc',
@@ -544,16 +526,6 @@ export const makeMessagesSocket = (config: SocketConfig) => {
return { nodes, shouldIncludeDeviceIdentity }
}
const createParticipantNodes = async (
recipientWireJids: string[],
message: proto.IMessage,
extraAttrs?: BinaryNode['attrs'],
dsmMessage?: proto.IMessage
) => {
const sessionMap = await resolveSessionJids(recipientWireJids)
return createParticipantNodesWithSessionMap(recipientWireJids, sessionMap, message, extraAttrs, dsmMessage)
}
const relayMessage = async (
jid: string,
message: proto.IMessage,
@@ -597,7 +569,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
const participants: BinaryNode[] = []
const destinationJid = !isStatus ? finalJid : statusJid
const binaryNodeContent: BinaryNode[] = []
const devices: DeviceWithWireJid[] = []
const devices: DeviceWithJid[] = []
const meMsg: proto.IMessage = {
deviceSentMessage: {
@@ -618,7 +590,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
devices.push({
user,
device,
wireJid: participant.jid
jid: participant.jid
})
}
@@ -718,10 +690,9 @@ export const makeMessagesSocket = (config: SocketConfig) => {
meId: groupSenderIdentity
})
const deviceSessionMap = await resolveSessionJids(devices.map(d => d.wireJid))
const senderKeyRecipients: string[] = []
for (const device of devices) {
const deviceJid = device.wireJid
const deviceJid = device.jid
const hasKey = !!senderKeyMap[deviceJid]
if (!hasKey && !isRetryResend) {
senderKeyRecipients.push(deviceJid)
@@ -739,15 +710,10 @@ export const makeMessagesSocket = (config: SocketConfig) => {
}
}
const senderKeySessionTargets = senderKeyRecipients.map(jid => deviceSessionMap.get(jid) ?? jid)
const senderKeySessionTargets = senderKeyRecipients
await assertSessions(senderKeySessionTargets)
const result = await createParticipantNodesWithSessionMap(
senderKeyRecipients,
deviceSessionMap,
senderKeyMsg,
extraAttrs
)
const result = await createParticipantNodes(senderKeyRecipients, senderKeyMsg, extraAttrs)
shouldIncludeDeviceIdentity = shouldIncludeDeviceIdentity || result.shouldIncludeDeviceIdentity
participants.push(...result.nodes)
@@ -785,7 +751,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
devices.push({
user,
device: 0,
wireJid: jidEncode(user, targetUserServer, 0)
jid: jidEncode(user, targetUserServer, 0)
})
// Own user matches conversation addressing mode
@@ -796,7 +762,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
devices.push({
user: ownUserForAddressing,
device: 0,
wireJid: jidEncode(ownUserForAddressing, ownUserServer, 0)
jid: jidEncode(ownUserForAddressing, ownUserServer, 0)
})
}
@@ -817,7 +783,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
logger.debug(
{
deviceCount: devices.length,
devices: devices.map(d => `${d.user}:${d.device}@${jidDecode(d.wireJid)?.server}`)
devices: devices.map(d => `${d.user}:${d.device}@${jidDecode(d.jid)?.server}`)
},
'Device enumeration complete with unified addressing'
)
@@ -830,10 +796,10 @@ export const makeMessagesSocket = (config: SocketConfig) => {
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)
for (const { user, jid } of devices) {
const isExactSenderDevice = jid === meId || (meLid && jid === meLid)
if (isExactSenderDevice) {
logger.debug({ wireJid, meId, meLid }, 'Skipping exact sender device (whatsmeow pattern)')
logger.debug({ jid, meId, meLid }, 'Skipping exact sender device (whatsmeow pattern)')
continue
}
@@ -841,25 +807,23 @@ export const makeMessagesSocket = (config: SocketConfig) => {
const isMe = user === mePnUser || (meLidUser && user === meLidUser)
if (isMe) {
meRecipients.push(wireJid)
meRecipients.push(jid)
} else {
otherRecipients.push(wireJid)
otherRecipients.push(jid)
}
allRecipients.push(wireJid)
allRecipients.push(jid)
}
const deviceSessionMap = await resolveSessionJids(devices.map(d => d.wireJid))
const sessionTargets = allRecipients.map(jid => deviceSessionMap.get(jid) ?? jid)
await assertSessions(sessionTargets)
await assertSessions(allRecipients)
const [
{ nodes: meNodes, shouldIncludeDeviceIdentity: s1 },
{ nodes: otherNodes, shouldIncludeDeviceIdentity: s2 }
] = await Promise.all([
// For own devices: use DSM if available (1:1 chats only)
createParticipantNodesWithSessionMap(meRecipients, deviceSessionMap, meMsg || message, extraAttrs),
createParticipantNodesWithSessionMap(otherRecipients, deviceSessionMap, message, extraAttrs, meMsg)
createParticipantNodes(meRecipients, meMsg || message, extraAttrs),
createParticipantNodes(otherRecipients, message, extraAttrs, meMsg)
])
participants.push(...meNodes)
participants.push(...otherNodes)