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 10a26a6..b8ae144 100644 Binary files a/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d and b/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d differ diff --git a/tests/artifacts/ios_backup/e7/e794f6ffcc3c222535f47684a63d5178da3c4500 b/tests/artifacts/ios_backup/e7/e794f6ffcc3c222535f47684a63d5178da3c4500 new file mode 100644 index 0000000..d54a9c5 Binary files /dev/null and b/tests/artifacts/ios_backup/e7/e794f6ffcc3c222535f47684a63d5178da3c4500 differ 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