From 06a949dafe3be11a3e5dd1b2f5ee21732e2e5617 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Donncha=20=C3=93=20Cearbhaill?= Date: Wed, 19 Aug 2026 13:06:41 +0200 Subject: [PATCH] Resolve WhatsApp LID chat identifiers via the LID pair table Recent WhatsApp versions key 1:1 chat sessions by an opaque LID rather than the contact's phone number. Extract the ZWAPHONENUMBERLIDPAIR table from the dedicated LID.sqlite database (or from ChatStorage itself in versions that store it there) and use it to populate partner_resolved_phone_number on chat session records and in timeline events, without requiring the often-missing ContactsV2.sqlite. Each pair is also extracted as a record and produces a lid_pair_recorded timeline event marking when the association was learned. --- docs/ios/records.md | 2 + src/mvt/ios/modules/mixed/whatsapp.py | 127 +++++++++++++++++- .../7c7fba66680ef796b916b067077cc246adacf01d | Bin 24576 -> 24576 bytes .../e794f6ffcc3c222535f47684a63d5178da3c4500 | Bin 0 -> 8192 bytes tests/ios_backup/test_whatsapp.py | 27 +++- tests/ios_fs/test_filesystem.py | 8 +- 6 files changed, 149 insertions(+), 15 deletions(-) create mode 100644 tests/artifacts/ios_backup/e7/e794f6ffcc3c222535f47684a63d5178da3c4500 diff --git a/docs/ios/records.md b/docs/ios/records.md index 909b627..b34183a 100644 --- a/docs/ios/records.md +++ b/docs/ios/records.md @@ -419,6 +419,8 @@ If indicators are provided through the command-line, they are checked against th This JSON file is created by mvt-ios' `WhatsApp` module. The module extracts a list of WhatsApp messages from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ChatStorage.sqlite*, along with one record per chat session (marked with `"record_type": "chat_session"`) containing the first and last interaction dates of each conversation. Chat sessions produce `chat_first_message` and `chat_last_message` timeline events, and group chats additionally produce a `group_created` event. A chat session's last-message date can postdate its newest stored message when the most recent messages in the chat were deleted. +Recent WhatsApp versions key 1:1 chat sessions by an opaque LID identifier rather than the contact's phone number. The module resolves these using the `ZWAPHONENUMBERLIDPAIR` table from the *LID.sqlite* database in the same app group (or from *ChatStorage.sqlite* itself in versions that store it there), populating `partner_resolved_phone_number` on chat session records and using the phone number in timeline events. Each LID-phone number pair is also extracted as a record (`"record_type": "lid_phone_number_pair"`) and produces a `lid_pair_recorded` timeline event marking when WhatsApp learned the association. + If indicators are provided through the command-line, they are checked against the extracted HTTP links. Any matches are stored in *whatsapp_detected.json*. --- diff --git a/src/mvt/ios/modules/mixed/whatsapp.py b/src/mvt/ios/modules/mixed/whatsapp.py index 73d8fba..89601df 100644 --- a/src/mvt/ios/modules/mixed/whatsapp.py +++ b/src/mvt/ios/modules/mixed/whatsapp.py @@ -4,8 +4,9 @@ # https://license.mvt.re/1.1/ import logging +import os import sqlite3 -from typing import Optional +from typing import Dict, Optional from mvt.common.module_types import ( ModuleAtomicResult, @@ -23,6 +24,26 @@ WHATSAPP_ROOT_PATHS = [ "private/var/mobile/Containers/Shared/AppGroup/*/ChatStorage.sqlite", ] +WHATSAPP_LID_BACKUP_IDS = [ + # SHA-1 of "AppDomainGroup-group.net.whatsapp.WhatsApp.shared-LID.sqlite" + "e794f6ffcc3c222535f47684a63d5178da3c4500", +] +WHATSAPP_LID_ROOT_PATHS = [ + "private/var/mobile/Containers/Shared/AppGroup/*/LID.sqlite", +] + +# WhatsApp records the mapping between a contact's LID and phone number +# identifiers in the ZWAPHONENUMBERLIDPAIR table. Depending on the WhatsApp +# version this lives in a dedicated LID.sqlite database or in +# ChatStorage.sqlite itself. +LID_PAIRS_QUERY = """ + SELECT + ZLID AS "lid", + ZPHONENUMBER AS "phone_number", + ZTIMESTAMP AS "pair_timestamp" + FROM ZWAPHONENUMBERLIDPAIR; +""" + CHAT_SESSIONS_QUERY = """ SELECT ZWACHATSESSION.Z_PK AS "session_pk", @@ -56,7 +77,8 @@ CHAT_SESSION_DATE_FIELDS = [ def _describe_chat(record: ModuleAtomicResult) -> str: jid = record.get("contact_jid") or "unknown" name = record.get("partner_name") - label = f"'{name}' ({jid})" if name else jid + identifier = record.get("partner_resolved_phone_number") or jid + label = f"'{name}' ({identifier})" if name else identifier is_group = jid.endswith("@g.us") or record.get("group_creation_date") if is_group: return f"WhatsApp group chat {label}" @@ -89,6 +111,18 @@ class Whatsapp(IOSExtraction): def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: if record.get("record_type") == "chat_session": return self._serialize_chat_session(record) + if record.get("record_type") == "lid_phone_number_pair": + if not record.get("pair_timestamp"): + return [] + return { + "timestamp": record["pair_timestamp"], + "module": self.__class__.__name__, + "event": "lid_pair_recorded", + "data": ( + f"WhatsApp associated LID {record.get('lid')} with " + f"phone number {record.get('phone_number')}" + ), + } text = record.get("ZTEXT", "").replace("\n", "\\n") links_text = "" @@ -235,18 +269,90 @@ class Whatsapp(IOSExtraction): self.results.append(message) total_messages = len(self.results) - total_sessions = self._extract_chat_sessions(cur) + lid_map = self._extract_lid_pairs(cur) + total_sessions = self._extract_chat_sessions(cur, lid_map) cur.close() conn.close() self.log.info( - "Extracted a total of %d WhatsApp messages and %d chat sessions", + "Extracted a total of %d WhatsApp messages, %d chat sessions " + "and %d LID-phone number pairs", total_messages, total_sessions, + len(lid_map), ) - def _extract_chat_sessions(self, cur: sqlite3.Cursor) -> int: + def _find_lid_db_path(self) -> Optional[str]: + for backup_id in WHATSAPP_LID_BACKUP_IDS: + file_path = self._get_backup_file_from_id(backup_id) + if file_path and os.path.exists(file_path): + return file_path + for found_path in self._get_fs_files_from_patterns( + WHATSAPP_LID_ROOT_PATHS + ): + return found_path + return None + + def _extract_lid_pairs(self, chat_cur: sqlite3.Cursor) -> Dict[str, str]: + """Extract the LID to phone number mapping from the dedicated + LID.sqlite database, falling back to the same table in + ChatStorage.sqlite. Returns a map of LID digits to phone number + digits.""" + rows = [] + lid_db_path = self._find_lid_db_path() + if lid_db_path: + self.log.info( + "Found WhatsApp LID database at path: %s", lid_db_path + ) + lid_conn = self._open_sqlite_db(lid_db_path) + try: + lid_cur = lid_conn.cursor() + lid_cur.execute(LID_PAIRS_QUERY) + rows = lid_cur.fetchall() + lid_cur.close() + except sqlite3.DatabaseError as exc: + self.log.warning( + "Unable to extract WhatsApp LID-phone number pairs: %s", + exc, + ) + finally: + lid_conn.close() + else: + try: + chat_cur.execute(LID_PAIRS_QUERY) + rows = chat_cur.fetchall() + except sqlite3.OperationalError: + self.log.info( + "No WhatsApp LID database found in this backup or " + "filesystem dump: LID chat identifiers cannot be " + "resolved to phone numbers" + ) + + lid_map: Dict[str, str] = {} + for lid, phone_number, pair_timestamp in rows: + record = { + "record_type": "lid_phone_number_pair", + "lid": lid, + "phone_number": phone_number, + "pair_timestamp": ( + convert_mactime_to_iso(pair_timestamp) or None + ) + if pair_timestamp + else None, + } + self.results.append(record) + + if lid and phone_number: + lid_digits = str(lid).split("@")[0] + phone_digits = str(phone_number).split("@")[0].lstrip("+") + lid_map[lid_digits] = phone_digits + + return lid_map + + def _extract_chat_sessions( + self, cur: sqlite3.Cursor, lid_map: Dict[str, str] + ) -> int: """Extract one record per chat session with the first and last interaction dates of each conversation.""" try: @@ -260,10 +366,19 @@ class Whatsapp(IOSExtraction): names = [description[0] for description in cur.description] total_sessions = 0 - for row in cur: + for row in cur.fetchall(): session = dict(zip(names, row)) session["record_type"] = "chat_session" + session["partner_resolved_phone_number"] = None + jid = session.get("contact_jid") or "" + if jid.endswith("@lid"): + phone_digits = lid_map.get(jid.split("@")[0]) + if phone_digits: + session["partner_resolved_phone_number"] = ( + f"+{phone_digits}" + ) + for field in CHAT_SESSION_DATE_FIELDS: if session.get(field): session[field] = ( diff --git a/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d b/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d index 10a26a644291b9d707edb56d490f8a05506470fa..b8ae144dd2db59a308c6f17793c47e861eeaee6d 100644 GIT binary patch delta 63 zcmZoTz}RqraRZwJ+YAQ&bNn+l3o7*RPoC&+reMIx&0ws~0R)UJ42A~Cz|bKlGiCET He<1+?BFzu` delta 48 zcmZoTz}RqraRZwJ+e8NbV&ETlc0R)UJ43i(oOKpDVFC+i} DVmS>- diff --git a/tests/artifacts/ios_backup/e7/e794f6ffcc3c222535f47684a63d5178da3c4500 b/tests/artifacts/ios_backup/e7/e794f6ffcc3c222535f47684a63d5178da3c4500 new file mode 100644 index 0000000000000000000000000000000000000000..d54a9c5554d55ba745082b0f28c1b9ab57705d40 GIT binary patch literal 8192 zcmeI#K}*9h6bJC64u`^+H?I!_aZJXF%C5`wp00Izz00bZa0SG_<0ub1sz~_go@ArE0tC7zhU*^+WWv)gU zXE*79ZK5xX(zdb~om*Cp$*DoYRM|pDyHsZSV1}eK$Cqr$Se+_6p&?I>gAUHzUrSgX znX0nWDUY2>4f$gvLBBsUIlgm2Dfpt|Rk3Gl^>4lq&BX@!u{8uB009U<00Izz00bZa i0SG_<0{;>?lyxC@YVCuq&b{u(-7s-{8lU-eefbxJO-!r+ literal 0 HcmV?d00001 diff --git a/tests/ios_backup/test_whatsapp.py b/tests/ios_backup/test_whatsapp.py index 0a62d0e..3bc920d 100644 --- a/tests/ios_backup/test_whatsapp.py +++ b/tests/ios_backup/test_whatsapp.py @@ -18,14 +18,24 @@ def test_extraction(): messages = [r for r in m.results if "ZTEXT" in r] sessions = [r for r in m.results if r.get("record_type") == "chat_session"] + pairs = [ + r for r in m.results + if r.get("record_type") == "lid_phone_number_pair" + ] assert len(messages) == 3 assert len(sessions) == 2 + assert len(pairs) == 1 + + assert pairs[0]["lid"] == "100000000000001" + assert pairs[0]["phone_number"] == "14155550100" + assert pairs[0]["pair_timestamp"] == "2025-08-25 07:33:20.000000" linked = next(r for r in messages if r.get("links")) assert linked["links"] == ["https://example.org/news"] alice = next(s for s in sessions if s["partner_name"] == "Alice Example") - assert alice["contact_jid"] == "14155550100@s.whatsapp.net" + assert alice["contact_jid"] == "100000000000001@lid" + assert alice["partner_resolved_phone_number"] == "+14155550100" assert alice["first_stored_message_date"] == "2025-08-27 15:06:40.000000" assert alice["last_message_date"] == "2025-08-28 18:53:20.000000" assert alice["group_creation_date"] is None @@ -39,19 +49,26 @@ def test_extraction(): assert group["last_stored_message_date"] == "2025-08-29 22:40:00.000000" assert group["last_message_date"] == "2025-08-31 02:26:40.000000" - # 3 message events plus first/last per chat and the group creation. - assert len(m.timeline) == 8 + # 3 message events, first/last per chat, the group creation and the + # LID-phone number pair. + assert len(m.timeline) == 9 events = { (entry["event"], entry["timestamp"]): entry["data"] for entry in m.timeline } + # Alice's session is keyed by LID but labelled with the phone number + # resolved through LID.sqlite. assert events[("chat_first_message", "2025-08-27 15:06:40.000000")] == ( "First stored message in WhatsApp chat with " - "'Alice Example' (14155550100@s.whatsapp.net)" + "'Alice Example' (+14155550100)" ) assert events[("chat_last_message", "2025-08-28 18:53:20.000000")] == ( "Last message in WhatsApp chat with " - "'Alice Example' (14155550100@s.whatsapp.net)" + "'Alice Example' (+14155550100)" + ) + assert events[("lid_pair_recorded", "2025-08-25 07:33:20.000000")] == ( + "WhatsApp associated LID 100000000000001 with " + "phone number 14155550100" ) assert events[("group_created", "2025-08-21 20:13:20.000000")] == ( "WhatsApp group chat 'Example Group' " diff --git a/tests/ios_fs/test_filesystem.py b/tests/ios_fs/test_filesystem.py index 3a7e0eb..636c004 100644 --- a/tests/ios_fs/test_filesystem.py +++ b/tests/ios_fs/test_filesystem.py @@ -15,8 +15,8 @@ class TestFilesystem: def test_filesystem(self): m = Filesystem(target_path=get_ios_backup_folder()) run_module(m) - assert len(m.results) == 21 - assert len(m.timeline) == 21 + assert len(m.results) == 23 + assert len(m.timeline) == 23 assert len(m.alertstore.alerts) == 0 def test_detection(self, indicator_file): @@ -29,6 +29,6 @@ class TestFilesystem: ) m.indicators = ind run_module(m) - assert len(m.results) == 21 - assert len(m.timeline) == 21 + assert len(m.results) == 23 + assert len(m.timeline) == 23 assert len(m.alertstore.alerts) == 1