feat(tctoken): complete lifecycle (TIER 1 + 2 + 3 of upstream PR)
feat(tctoken): complete lifecycle (TIER 1 + 2 + 3 of upstream PR)
This commit is contained in:
@@ -57,6 +57,16 @@ export type IdentityChangeContext = {
|
||||
debounceCache: NodeCache<boolean>
|
||||
/** Logger instance for debugging and monitoring */
|
||||
logger: ILogger
|
||||
/**
|
||||
* Invoked right before `assertSessions` is called for an existing-session identity
|
||||
* change. Used to kick off fire-and-forget side effects (e.g. tctoken re-issuance)
|
||||
* in the same order WA Web does — i.e. before the E2E session is re-established.
|
||||
* Must not throw; implementations are responsible for their own error handling.
|
||||
*
|
||||
* Skipped when the refresh itself is skipped (no_identity_node, invalid_notification,
|
||||
* skipped_companion_device, skipped_self_primary, debounced, skipped_offline).
|
||||
*/
|
||||
onBeforeSessionRefresh?: (jid: string) => void
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
@@ -170,6 +180,19 @@ export async function handleIdentityChange(
|
||||
// This ensures we don't incorrectly debounce when we exit early (offline, etc.)
|
||||
ctx.debounceCache.set(from, true)
|
||||
|
||||
// Fire-and-forget side effects (e.g. tctoken re-issuance) BEFORE the session is
|
||||
// re-established. WA Web runs these in parallel with the session refresh —
|
||||
// running afterwards would race with the next outbound send and risk error 463.
|
||||
//
|
||||
// Wrapped in try/catch so a misbehaving consumer callback cannot abort identity
|
||||
// change recovery. We log and continue — assertSessions still runs so the E2E
|
||||
// session always gets refreshed.
|
||||
try {
|
||||
ctx.onBeforeSessionRefresh?.(from)
|
||||
} catch (error) {
|
||||
ctx.logger.warn({ error, jid: from }, 'onBeforeSessionRefresh callback threw — continuing with session refresh')
|
||||
}
|
||||
|
||||
// Attempt session refresh/creation
|
||||
try {
|
||||
await ctx.assertSessions([from], true)
|
||||
|
||||
@@ -40,6 +40,7 @@ import { getKeyAuthor, toNumber } from './generics'
|
||||
import { downloadAndProcessHistorySyncNotification } from './history'
|
||||
import type { ILogger } from './logger'
|
||||
import { metrics, recordHistorySyncMessages } from './prometheus-metrics.js'
|
||||
import { buildMergedTcTokenIndexWrite, resolveTcTokenJid } from './tc-token-utils'
|
||||
|
||||
type ProcessMessageContext = {
|
||||
shouldProcessHistoryMsg: boolean
|
||||
@@ -62,6 +63,89 @@ const REAL_MSG_STUB_TYPES = new Set([
|
||||
|
||||
const REAL_MSG_REQ_ME_STUB_TYPES = new Set([WAMessageStubType.GROUP_PARTICIPANT_ADD])
|
||||
|
||||
/**
|
||||
* Extract tctoken / tcTokenTimestamp / tcTokenSenderTimestamp from history-sync chats
|
||||
* and persist them to the `tctoken` store. Mirrors WA Web's `bulkCreateOrMerge` pass
|
||||
* over the chat table during history sync.
|
||||
*
|
||||
* Why this matters: when a user logs in on a new device, the multi-device history sync
|
||||
* is the only way that device learns about tctokens issued/received on the original
|
||||
* device. Without this pass, the new device sends 1:1 messages with no tctoken until
|
||||
* the contact triggers a fresh notification — which surfaces as error 463 in production.
|
||||
*
|
||||
* Monotonicity: we only overwrite an existing entry if the incoming timestamp is
|
||||
* STRICTLY newer (`incoming > existing`). Equal timestamps are skipped to avoid
|
||||
* reverting senderTimestamp / realIssueTimestamp set by other layers (e.g. a
|
||||
* reissue that fired between history-sync chunks).
|
||||
*
|
||||
* Index hygiene: every JID we write here is added to the persistent prune index
|
||||
* (TC_TOKEN_INDEX_KEY) via buildMergedTcTokenIndexWrite, so the 24h prune sweep in
|
||||
* messages-recv picks them up across sessions.
|
||||
*/
|
||||
async function storeTcTokensFromHistorySync(
|
||||
chats: Chat[],
|
||||
signalRepository: SignalRepositoryWithLIDStore,
|
||||
keyStore: SignalKeyStoreWithTransaction,
|
||||
logger?: ILogger
|
||||
) {
|
||||
const getLIDForPN = signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping)
|
||||
|
||||
const candidates: { storageJid: string; token: Buffer; ts: number; senderTs?: number }[] = []
|
||||
for (const chat of chats) {
|
||||
const ts = chat.tcTokenTimestamp ? toNumber(chat.tcTokenTimestamp) : 0
|
||||
if (chat.tcToken?.length && ts > 0) {
|
||||
const jid = jidNormalizedUser(chat.id!)
|
||||
const storageJid = await resolveTcTokenJid(jid, getLIDForPN)
|
||||
candidates.push({
|
||||
storageJid,
|
||||
token: Buffer.from(chat.tcToken),
|
||||
ts,
|
||||
senderTs: chat.tcTokenSenderTimestamp ? toNumber(chat.tcTokenSenderTimestamp) : undefined
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
if (!candidates.length) {
|
||||
return
|
||||
}
|
||||
|
||||
const jids = candidates.map(c => c.storageJid)
|
||||
const existing = await keyStore.get('tctoken', jids)
|
||||
const entries: Record<string, { token: Buffer; timestamp?: string; senderTimestamp?: number }> = {}
|
||||
|
||||
for (const c of candidates) {
|
||||
// Same-batch dedup: when two chats resolve to the same storageJid (e.g. PN+LID
|
||||
// aliases collapsing through resolveTcTokenJid, or duplicate chunks across
|
||||
// retries), prefer the value already written by an earlier iteration so a
|
||||
// lower-ts entry can't overwrite a higher-ts one captured from `existing`.
|
||||
const existingEntry = entries[c.storageJid] ?? existing[c.storageJid]
|
||||
const existingTs = existingEntry?.timestamp ? Number(existingEntry.timestamp) : 0
|
||||
// Strict > guard: equal timestamps are skipped so we never clobber
|
||||
// senderTimestamp written by other layers (issuance after send, etc).
|
||||
if (existingTs > 0 && existingTs >= c.ts) {
|
||||
continue
|
||||
}
|
||||
|
||||
entries[c.storageJid] = {
|
||||
...existingEntry,
|
||||
token: c.token,
|
||||
timestamp: String(c.ts),
|
||||
...(c.senderTs !== undefined ? { senderTimestamp: c.senderTs } : {})
|
||||
}
|
||||
}
|
||||
|
||||
if (Object.keys(entries).length) {
|
||||
logger?.debug({ count: Object.keys(entries).length }, 'storing tctokens from history sync')
|
||||
try {
|
||||
// Include updated __index so cross-session pruning picks these JIDs up.
|
||||
const indexWrite = await buildMergedTcTokenIndexWrite(keyStore, Object.keys(entries))
|
||||
await keyStore.set({ tctoken: { ...entries, ...indexWrite } })
|
||||
} catch (err) {
|
||||
logger?.warn({ err }, 'failed to store tctokens from history sync')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Cleans a received message to further processing */
|
||||
export const cleanMessage = (message: WAMessage, meId: string, meLid: string) => {
|
||||
// ensure remoteJid and participant doesn't have device or agent in it
|
||||
@@ -418,6 +502,12 @@ const processMessage = async (
|
||||
|
||||
// Emit LID-PN mappings from history sync
|
||||
// This is how WhatsApp Web learns mappings for chats with non-contacts
|
||||
//
|
||||
// MUST run BEFORE storeTcTokensFromHistorySync — otherwise resolveTcTokenJid()
|
||||
// can't resolve PN→LID for fresh-device chats (mapping cache is empty), tokens
|
||||
// get persisted under PN keys, and the send path (which resolves to LID first)
|
||||
// misses them — exactly the error 463 scenario this whole change is meant to
|
||||
// prevent. Catches Codex P1 review on PR #386.
|
||||
if (data.lidPnMappings?.length) {
|
||||
logger?.debug({ count: data.lidPnMappings.length }, 'processing LID-PN mappings from history sync')
|
||||
// eslint-disable-next-line max-depth
|
||||
@@ -442,6 +532,12 @@ const processMessage = async (
|
||||
}
|
||||
}
|
||||
|
||||
// Persist tctokens carried by history-sync chats BEFORE emitting messaging-history.set
|
||||
// — listeners may immediately fire outbound sends that need the tctoken, and the store
|
||||
// has to be populated first to avoid an error 463 on the first multi-device send.
|
||||
// Runs AFTER storeLIDPNMappings (see comment above) so LID resolution works.
|
||||
await storeTcTokensFromHistorySync(data.chats, signalRepository, keyStore, logger)
|
||||
|
||||
ev.emit('messaging-history.set', {
|
||||
...data,
|
||||
isLatest: histNotification.syncType !== proto.HistorySync.HistorySyncType.ON_DEMAND ? isLatest : undefined,
|
||||
|
||||
+119
-2
@@ -1,12 +1,86 @@
|
||||
import type { SignalKeyStoreWithTransaction } from '../Types'
|
||||
import type { BinaryNode } from '../WABinary'
|
||||
import { getBinaryNodeChild, getBinaryNodeChildren, isLidUser, jidNormalizedUser } from '../WABinary'
|
||||
import {
|
||||
getBinaryNodeChild,
|
||||
getBinaryNodeChildren,
|
||||
isAnyLidUser,
|
||||
isAnyPnUser,
|
||||
isHostedLidUser,
|
||||
isHostedPnUser,
|
||||
isJidMetaAI,
|
||||
isLidUser,
|
||||
isPnUser,
|
||||
jidNormalizedUser
|
||||
} from '../WABinary'
|
||||
|
||||
/** 7 days in seconds — matches WA Web AB prop tctoken_duration */
|
||||
const TC_TOKEN_BUCKET_DURATION = 604800
|
||||
/** 4 buckets → ~28-day rolling window — matches WA Web AB prop tctoken_num_buckets */
|
||||
const TC_TOKEN_NUM_BUCKETS = 4
|
||||
|
||||
/**
|
||||
* Sentinel key under the `tctoken` store holding a JSON array of tracked storage JIDs
|
||||
* for cross-session pruning. Mirrors WA Web's CLEAN_TC_TOKENS index lookup.
|
||||
*
|
||||
* Exported so other modules (messages-recv, messages-send, process-message) reference
|
||||
* the same constant. The fork previously inlined this string in messages-recv.ts;
|
||||
* keep the value identical (`'__index'`) for backward compatibility with persisted state.
|
||||
*/
|
||||
export const TC_TOKEN_INDEX_KEY = '__index'
|
||||
|
||||
// Phone-number pattern matching WABinary's isJidBot, applied against the user part so
|
||||
// the check is invariant to @c.us ↔ @s.whatsapp.net normalization.
|
||||
const BOT_PHONE_REGEX = /^1313555\d{4}$|^131655500\d{2}$/
|
||||
|
||||
/**
|
||||
* Mirrors WA Web's `Wid.isRegularUser()` (user ∧ ¬PSA ∧ ¬Bot). Used to gate tctoken
|
||||
* storage against malformed notifications — WA Web filters server-side but we
|
||||
* defend here for parity with `WAWebSetTcTokenChatAction.handleIncomingTcToken`.
|
||||
* Works for both pre- and post-normalized JIDs (`@c.us` vs `@s.whatsapp.net`).
|
||||
*/
|
||||
export function isRegularUser(jid: string | undefined): boolean {
|
||||
if (!jid) return false
|
||||
const user = jid.split('@')[0] ?? ''
|
||||
if (user === '0') return false // PSA
|
||||
if (BOT_PHONE_REGEX.test(user)) return false // Bot by phone pattern
|
||||
if (isJidMetaAI(jid)) return false // MetaAI (@bot server)
|
||||
return !!(isPnUser(jid) || isLidUser(jid) || isHostedPnUser(jid) || isHostedLidUser(jid) || jid.endsWith('@c.us'))
|
||||
}
|
||||
|
||||
/** Read the persisted tctoken JID index and return its entries (never contains the sentinel key itself). */
|
||||
export async function readTcTokenIndex(keys: SignalKeyStoreWithTransaction): Promise<string[]> {
|
||||
const data = await keys.get('tctoken', [TC_TOKEN_INDEX_KEY])
|
||||
const entry = data[TC_TOKEN_INDEX_KEY]
|
||||
if (!entry?.token?.length) return []
|
||||
try {
|
||||
const parsed = JSON.parse(Buffer.from(entry.token).toString())
|
||||
if (!Array.isArray(parsed)) return []
|
||||
return parsed.filter((j): j is string => typeof j === 'string' && j.length > 0 && j !== TC_TOKEN_INDEX_KEY)
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a SignalDataSet fragment that writes the merged index (persisted ∪ added)
|
||||
* under the sentinel key. Lets callers update the index without clobbering writes
|
||||
* made by other layers (history sync, concurrent sessions on the same store).
|
||||
*/
|
||||
export async function buildMergedTcTokenIndexWrite(
|
||||
keys: SignalKeyStoreWithTransaction,
|
||||
addedJids: Iterable<string>
|
||||
): Promise<{ [TC_TOKEN_INDEX_KEY]: { token: Buffer } }> {
|
||||
const persisted = await readTcTokenIndex(keys)
|
||||
const merged = new Set(persisted)
|
||||
for (const jid of addedJids) {
|
||||
if (jid && jid !== TC_TOKEN_INDEX_KEY) merged.add(jid)
|
||||
}
|
||||
|
||||
return {
|
||||
[TC_TOKEN_INDEX_KEY]: { token: Buffer.from(JSON.stringify([...merged])) }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a received token is expired using WA Web's rolling bucket algorithm.
|
||||
* Reference: WAWebTrustedContactsUtils.isTokenExpired
|
||||
@@ -63,6 +137,43 @@ export async function resolveTcTokenJid(
|
||||
return lid ?? normalized
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve target JID for issuing privacy token based on AB prop 14303
|
||||
* (`lid_trusted_token_issue_to_lid`). When the prop is on, issuance goes to
|
||||
* the LID; when off, it goes to the PN. Returns the original JID if no
|
||||
* mapping is found in either direction.
|
||||
*
|
||||
* Normalizes the JID upfront and uses the `isAny*` helpers so callers can pass
|
||||
* `@c.us`, `@s.whatsapp.net`, `@hosted`, `@hosted.lid`, `@lid` or device-specific
|
||||
* forms — `LIDMappingStore.getLIDForPN` early-returns unless `isAnyPnUser`,
|
||||
* so unnormalized inputs would silently bypass routing.
|
||||
*
|
||||
* Reference: WAWebTrustedContactsManager.issuePrivacyTokens
|
||||
*/
|
||||
export async function resolveIssuanceJid(
|
||||
jid: string,
|
||||
issueToLid: boolean,
|
||||
getLIDForPN: (pn: string) => Promise<string | null>,
|
||||
getPNForLID?: (lid: string) => Promise<string | null>
|
||||
): Promise<string> {
|
||||
const normalized = jidNormalizedUser(jid)
|
||||
|
||||
if (issueToLid) {
|
||||
if (isAnyLidUser(normalized)) return normalized
|
||||
if (!isAnyPnUser(normalized)) return normalized
|
||||
const lid = await getLIDForPN(normalized)
|
||||
return lid ?? normalized
|
||||
}
|
||||
|
||||
if (!isAnyLidUser(normalized)) return normalized
|
||||
if (getPNForLID) {
|
||||
const pn = await getPNForLID(normalized)
|
||||
return pn ?? normalized
|
||||
}
|
||||
|
||||
return normalized
|
||||
}
|
||||
|
||||
type TcTokenParams = {
|
||||
jid: string
|
||||
baseContent?: BinaryNode[]
|
||||
@@ -135,7 +246,13 @@ export async function storeTcTokensFromIqResult({
|
||||
continue
|
||||
}
|
||||
|
||||
const rawJid = jidNormalizedUser(tokenNode.attrs.jid || fallbackJid)
|
||||
// In notifications, tokenNode.attrs.jid is OUR own device JID, not the sender's.
|
||||
// Prefer fallbackJid (resolved from notification's `from` / `sender_lid`) so the
|
||||
// token is stored under the peer's JID, never under self.
|
||||
const rawJid = jidNormalizedUser(fallbackJid || tokenNode.attrs.jid)
|
||||
// Defense against malformed notifications (PSA WID '0', bots, MetaAI). WA Web
|
||||
// filters these server-side; we mirror Wid.isRegularUser() locally.
|
||||
if (!isRegularUser(rawJid)) continue
|
||||
const storageJid = await resolveTcTokenJid(rawJid, getLIDForPN)
|
||||
const existingTcData = await keys.get('tctoken', [storageJid])
|
||||
const existingEntry = existingTcData[storageJid]
|
||||
|
||||
Reference in New Issue
Block a user