diff --git a/docs/ios/records.md b/docs/ios/records.md index 016c861..b34183a 100644 --- a/docs/ios/records.md +++ b/docs/ios/records.md @@ -417,7 +417,21 @@ If indicators are provided through the command-line, they are checked against th Backup: :material-check: Full filesystem dump: :material-check: -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*. +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*. +--- + +### `whatsapp_contacts.json` + +!!! info "Availability" + Backup: :material-check: + Full filesystem dump: :material-check: + +This JSON file is created by mvt-ios' `WhatsappContacts` module. The module extracts WhatsApp contact records from the SQLite database located at *private/var/mobile/Containers/Shared/AppGroup/\*/ContactsV2.sqlite*, including each contact's phone number, WhatsApp and LID identifiers, and the per-contact disappearing messages timer, which is not recorded in *ChatStorage.sqlite*. Each timestamp stored on a contact record produces a timeline event: `disappearing_mode_set` (when the disappearing messages timer was last changed), `about_changed` (when the contact last changed their "about" text), `about_expiration` (when a timed "about" is scheduled to expire) and `contact_last_updated` (when the contact record was last updated). + +This database is often missing from incremental backups. When it cannot be found, the module logs a warning and produces no results, in which case the disappearing messages state of chats cannot be determined from the backup. + diff --git a/src/mvt/ios/modules/mixed/__init__.py b/src/mvt/ios/modules/mixed/__init__.py index 0e50c26..a846bf2 100644 --- a/src/mvt/ios/modules/mixed/__init__.py +++ b/src/mvt/ios/modules/mixed/__init__.py @@ -26,6 +26,7 @@ from .tcc import TCC from .webkit_resource_load_statistics import WebkitResourceLoadStatistics from .webkit_session_resource_log import WebkitSessionResourceLog from .whatsapp import Whatsapp +from .whatsapp_contacts import WhatsappContacts MIXED_MODULES = [ Calls, @@ -47,6 +48,7 @@ MIXED_MODULES = [ WebkitResourceLoadStatistics, WebkitSessionResourceLog, Whatsapp, + WhatsappContacts, Shortcuts, Applications, Calendar, diff --git a/src/mvt/ios/modules/mixed/interactionc.py b/src/mvt/ios/modules/mixed/interactionc.py index 81a67e2..4d21382 100644 --- a/src/mvt/ios/modules/mixed/interactionc.py +++ b/src/mvt/ios/modules/mixed/interactionc.py @@ -3,9 +3,11 @@ # Use of this software is governed by the MVT License 1.1 that can be found at # https://license.mvt.re/1.1/ +import datetime import logging +import re import sqlite3 -from typing import Optional +from typing import Optional, Tuple from mvt.common.module_types import ( ModuleAtomicResult, @@ -15,6 +17,7 @@ from mvt.common.module_types import ( from mvt.common.utils import convert_mactime_to_iso from ..base import IOSExtraction +from .whatsapp_contacts import WhatsappContacts INTERACTIONC_BACKUP_IDS = [ "1f5a521220a3ad80ebfdc196978df8e7a2e49dee", @@ -22,6 +25,49 @@ INTERACTIONC_BACKUP_IDS = [ INTERACTIONC_ROOT_PATHS = [ "private/var/mobile/Library/CoreDuet/People/interactionC.db", ] + +# The interaction record's creation date normally trails its start date by +# milliseconds: emitting it as a timeline event only duplicates the start +# date event. A large divergence, however, indicates the record was +# backfilled (sync, restore, or tampering) and is worth surfacing. +CREATION_DATE_DIVERGENCE_THRESHOLD = 3600.0 + +# Per-contact aggregate dates from ZCONTACTS are repeated on every +# interaction row of the same contact. They are serialized with a +# contact-centric data string so that timeline de-duplication collapses +# them into one event per contact. +CONTACT_EVENT_TEMPLATES = { + "contacts_creation_date": "Contact {party} first recorded in interactionC", + "first_incoming_sender_date": "First incoming interaction from {party}", + "last_incoming_sender_date": "Last incoming interaction from {party}", + "first_incoming_recipient_date": ( + "First incoming interaction where {party} was a recipient" + ), + "last_incoming_recipient_date": ( + "Last incoming interaction where {party} was a recipient" + ), + "first_outgoing_recipient_date": ( + "First outgoing interaction to {party}" + ), + "last_outgoing_recipient_date": ( + "Last outgoing interaction to {party}" + ), +} + + +def _parse_iso(timestamp) -> Optional[datetime.datetime]: + try: + return datetime.datetime.strptime( + timestamp, "%Y-%m-%d %H:%M:%S.%f" + ) + except (TypeError, ValueError): + return None + + +def _describe_delta(seconds: float) -> str: + if seconds >= 86400: + return f"{seconds / 86400:.0f} days" + return f"{seconds / 3600:.0f} hours" # Taken from APOLLO # https://github.com/mac4n6/APOLLO/blob/master/modules/interaction_contact_interactions.txt QUERIES = [ @@ -34,7 +80,7 @@ QUERIES = [ CASE ZINTERACTIONS.ZDIRECTION WHEN '0' THEN 'INCOMING' WHEN '1' THEN 'OUTGOING' - END 'DIRECTION' AS "direction", + END AS "direction", ZCONTACTS.ZDISPLAYNAME AS "sender_display_name", ZCONTACTS.ZIDENTIFIER AS "sender_identifier", ZCONTACTS.ZPERSONID AS "sender_personid", @@ -89,7 +135,7 @@ QUERIES = [ CASE ZINTERACTIONS.ZDIRECTION WHEN '0' THEN 'INCOMING' WHEN '1' THEN 'OUTGOING' - END 'DIRECTION' AS "direction", + END AS "direction", ZCONTACTS.ZDISPLAYNAME AS "sender_display_name", ZCONTACTS.ZIDENTIFIER AS "sender_identifier", ZCONTACTS.ZPERSONID AS "sender_personid", @@ -117,7 +163,7 @@ QUERIES = [ CASE ZCONTACTS.ZLASTINCOMINGRECIPIENTDATE WHEN '0' THEN '0' ELSE ZCONTACTS.ZLASTINCOMINGRECIPIENTDATE - END 'LAST INCOMING RECIPIENT DATE' AS "last_incoming_recipient_date", + END AS "last_incoming_recipient_date", ZCONTACTS.ZLASTOUTGOINGRECIPIENTDATE AS "last_outgoing_recipient_date", ZCONTACTS.ZCUSTOMIDENTIFIER AS "custom_id", ZINTERACTIONS.ZCONTENTURL AS "interaction_content_url", @@ -218,9 +264,16 @@ QUERIES = [ ] +WHATSAPP_BUNDLE_ID = "net.whatsapp.WhatsApp" + + class InteractionC(IOSExtraction): """This module extracts data from InteractionC db.""" + # WhatsApp identifies chat peers by LID in interactionC.db, which only the + # WhatsApp contacts database can map back to a phone number and name. + dependencies = [WhatsappContacts] + def __init__( self, file_path: Optional[str] = None, @@ -252,10 +305,51 @@ class InteractionC(IOSExtraction): "last_outgoing_recipient_date", ] + @staticmethod + def _describe_party(record: ModuleAtomicResult, prefix: str) -> Optional[str]: + name = record.get(f"{prefix}_display_name") or record.get( + f"{prefix}_resolved_name" + ) + identifier = record.get(f"{prefix}_resolved_phone_number") or record.get( + f"{prefix}_identifier" + ) + if name and identifier: + # A display name that is just a formatted copy of the phone + # number adds no information. + name_digits = re.sub(r"\D", "", name) + if name_digits and name_digits == re.sub(r"\D", "", identifier): + return identifier + return f"{name} ({identifier})" + return name or identifier or None + def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: + sender = self._describe_party(record, "sender") + # The chat peer from the domain identifier stands in when the + # recipient was not recorded (or the recipient join is unavailable). + recipient = self._describe_party(record, "recipient") or self._describe_party( + record, "domain" + ) + direction = record.get("direction") + if not sender and direction == "OUTGOING": + sender = "local user" + if not recipient and direction == "INCOMING": + recipient = "local user" + + header = f"[{record['bundle_id']}]" + if record.get("account"): + header += f" {record['account']}" + if direction: + header += f" {direction}" + + data = f"{header} from {sender or 'unknown'} to {recipient or 'unknown'}" + if record.get("group_name"): + data += f" (group: {record['group_name']})" + if record.get("content"): + data += f": {record['content']}" + records = [] processed = [] - for timestamp in self.timestamps: + for timestamp in ("start_date", "end_date"): # Check if the record has the current timestamp. if timestamp not in record or not record[timestamp]: continue @@ -269,16 +363,142 @@ class InteractionC(IOSExtraction): "timestamp": record[timestamp], "module": self.__class__.__name__, "event": timestamp, - "data": f"[{record['bundle_id']}] {record['account']} - " - f"from {record['sender_display_name']} ({record['sender_identifier']}) " - f"to {record.get('recipient_display_name', '')} ({record.get('recipient_identifier', '')}):" - f" {record.get('content', '')}", + "data": data, } ) processed.append(record[timestamp]) + creation_event = self._serialize_creation_date(record, data) + if creation_event: + records.append(creation_event) + + # Contact-level aggregates describe the sender's contact record. + party = self._describe_party(record, "sender") + if party: + for field, template in CONTACT_EVENT_TEMPLATES.items(): + if not record.get(field): + continue + records.append( + { + "timestamp": record[field], + "module": self.__class__.__name__, + "event": field, + "data": template.format(party=party), + } + ) + return records + def _serialize_creation_date( + self, record: ModuleAtomicResult, data: str + ) -> Optional[dict]: + """Serialize the interaction record's creation date only when it + diverges from the start date enough to indicate the record was + backfilled.""" + creation = record.get("interactions_creation_date") + if not creation: + return None + + event = { + "timestamp": creation, + "module": self.__class__.__name__, + "event": "interactions_creation_date", + "data": data, + } + + start = _parse_iso(record.get("start_date")) + creation_parsed = _parse_iso(creation) + if not start or not creation_parsed: + # Without a start date the creation date is the only anchor. + return event + + delta = (creation_parsed - start).total_seconds() + if abs(delta) < CREATION_DATE_DIVERGENCE_THRESHOLD: + return None + + direction = "after" if delta > 0 else "before" + event["data"] = ( + f"Interaction record created {_describe_delta(abs(delta))} " + f"{direction} the event: {data}" + ) + return event + + def _whatsapp_contact_maps(self) -> Tuple[dict, dict]: + """Build LID and phone-digit lookup maps from the WhatsappContacts + module results, when available.""" + by_lid: dict = {} + by_phone: dict = {} + contacts_module = self.dependency_modules.get(WhatsappContacts) + if not contacts_module: + return by_lid, by_phone + + for contact in contacts_module.results: + name = contact.get("full_name") or contact.get("given_name") + phone = contact.get("phone_number") + entry = (phone, name) + if contact.get("lid"): + by_lid[contact["lid"]] = entry + if phone: + by_phone[re.sub(r"\D", "", phone)] = entry + whatsapp_id = contact.get("whatsapp_id") + if whatsapp_id and "@" in whatsapp_id: + by_phone.setdefault(whatsapp_id.split("@")[0], entry) + + return by_lid, by_phone + + @staticmethod + def _resolve_whatsapp_identifier( + value, by_lid: dict, by_phone: dict + ) -> Tuple[Optional[str], Optional[str]]: + """Resolve a WhatsApp identifier (LID, JID or phone number) to a + (phone_number, contact_name) tuple.""" + if not value: + return None, None + + value = str(value) + if value.endswith("@lid"): + return by_lid.get(value, (None, None)) + if value.endswith("@g.us"): + return None, None + if value.endswith("@s.whatsapp.net"): + digits = value.split("@")[0] + phone, name = by_phone.get(digits, (None, None)) + return phone or f"+{digits}", name + if value.startswith("+"): + _, name = by_phone.get(re.sub(r"\D", "", value), (None, None)) + return None, name + + return None, None + + def _postprocess_results(self) -> None: + by_lid, by_phone = self._whatsapp_contact_maps() + + for entry in self.results: + # The fallback queries return ZDIRECTION raw instead of labelled. + if entry.get("direction") in (0, "0"): + entry["direction"] = "INCOMING" + elif entry.get("direction") in (1, "1"): + entry["direction"] = "OUTGOING" + + if entry.get("bundle_id") != WHATSAPP_BUNDLE_ID: + continue + + candidates = { + "sender": [entry.get("sender_identifier"), entry.get("custom_id")], + "recipient": [entry.get("recipient_identifier")], + "domain": [entry.get("domain_identifier")], + } + for prefix, values in candidates.items(): + phone = name = None + for value in values: + phone, name = self._resolve_whatsapp_identifier( + value, by_lid, by_phone + ) + if phone or name: + break + entry[f"{prefix}_resolved_phone_number"] = phone + entry[f"{prefix}_resolved_name"] = name + def run(self) -> None: self._find_ios_database( backup_ids=INTERACTIONC_BACKUP_IDS, root_paths=INTERACTIONC_ROOT_PATHS @@ -325,4 +545,6 @@ class InteractionC(IOSExtraction): cur.close() conn.close() + self._postprocess_results() + self.log.info("Extracted a total of %d InteractionC events", len(self.results)) diff --git a/src/mvt/ios/modules/mixed/whatsapp.py b/src/mvt/ios/modules/mixed/whatsapp.py index 0a80aad..89601df 100644 --- a/src/mvt/ios/modules/mixed/whatsapp.py +++ b/src/mvt/ios/modules/mixed/whatsapp.py @@ -4,7 +4,9 @@ # https://license.mvt.re/1.1/ import logging -from typing import Optional +import os +import sqlite3 +from typing import Dict, Optional from mvt.common.module_types import ( ModuleAtomicResult, @@ -22,9 +24,71 @@ 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", + ZWACHATSESSION.ZCONTACTJID AS "contact_jid", + ZWACHATSESSION.ZPARTNERNAME AS "partner_name", + ZWACHATSESSION.ZSESSIONTYPE AS "session_type", + ZWACHATSESSION.ZARCHIVED AS "archived", + ZWACHATSESSION.ZREMOVED AS "removed", + ZWACHATSESSION.ZMESSAGECOUNTER AS "message_counter", + ZWACHATSESSION.ZLASTMESSAGEDATE AS "last_message_date", + ZWAGROUPINFO.ZCREATIONDATE AS "group_creation_date", + MIN(ZWAMESSAGE.ZMESSAGEDATE) AS "first_stored_message_date", + MAX(ZWAMESSAGE.ZMESSAGEDATE) AS "last_stored_message_date", + COUNT(ZWAMESSAGE.Z_PK) AS "stored_message_count" + FROM ZWACHATSESSION + LEFT JOIN ZWAGROUPINFO + ON ZWACHATSESSION.ZGROUPINFO = ZWAGROUPINFO.Z_PK + LEFT JOIN ZWAMESSAGE + ON ZWAMESSAGE.ZCHATSESSION = ZWACHATSESSION.Z_PK + GROUP BY ZWACHATSESSION.Z_PK; +""" + +CHAT_SESSION_DATE_FIELDS = [ + "last_message_date", + "group_creation_date", + "first_stored_message_date", + "last_stored_message_date", +] + + +def _describe_chat(record: ModuleAtomicResult) -> str: + jid = record.get("contact_jid") or "unknown" + name = record.get("partner_name") + 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}" + return f"WhatsApp chat with {label}" + class Whatsapp(IOSExtraction): - """This module extracts all WhatsApp messages containing links.""" + """This module extracts all WhatsApp messages containing links, as well + as per-chat records with the first and last interaction dates of each + conversation.""" def __init__( self, @@ -45,6 +109,21 @@ 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 = "" if record.get("links"): @@ -57,6 +136,49 @@ class Whatsapp(IOSExtraction): "data": f"'{text}' from {record.get('ZFROMJID', 'Unknown')}{links_text}", } + def _serialize_chat_session( + self, record: ModuleAtomicResult + ) -> ModuleSerializedResult: + records = [] + chat = _describe_chat(record) + + if record.get("group_creation_date"): + records.append( + { + "timestamp": record["group_creation_date"], + "module": self.__class__.__name__, + "event": "group_created", + "data": f"{chat} was created", + } + ) + + if record.get("first_stored_message_date"): + records.append( + { + "timestamp": record["first_stored_message_date"], + "module": self.__class__.__name__, + "event": "chat_first_message", + "data": f"First stored message in {chat}", + } + ) + + # The chat session's own last-message date is authoritative: it can + # postdate the newest stored message if that message was deleted. + last_message_date = record.get("last_message_date") or record.get( + "last_stored_message_date" + ) + if last_message_date: + records.append( + { + "timestamp": last_message_date, + "module": self.__class__.__name__, + "event": "chat_last_message", + "data": f"Last message in {chat}", + } + ) + + return records + def check_indicators(self) -> None: if not self.indicators: return @@ -146,7 +268,126 @@ class Whatsapp(IOSExtraction): message["links"] = list(set(filtered_links)) self.results.append(message) + total_messages = len(self.results) + 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", len(self.results)) + self.log.info( + "Extracted a total of %d WhatsApp messages, %d chat sessions " + "and %d LID-phone number pairs", + total_messages, + total_sessions, + len(lid_map), + ) + + 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: + cur.execute(CHAT_SESSIONS_QUERY) + except sqlite3.OperationalError as exc: + self.log.warning( + "Unable to extract WhatsApp chat sessions: %s", exc + ) + return 0 + + names = [description[0] for description in cur.description] + + total_sessions = 0 + 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] = ( + convert_mactime_to_iso(session[field]) or None + ) + else: + session[field] = None + + self.results.append(session) + total_sessions += 1 + + return total_sessions diff --git a/src/mvt/ios/modules/mixed/whatsapp_contacts.py b/src/mvt/ios/modules/mixed/whatsapp_contacts.py new file mode 100644 index 0000000..309da22 --- /dev/null +++ b/src/mvt/ios/modules/mixed/whatsapp_contacts.py @@ -0,0 +1,308 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 The MVT Authors. +# Use of this software is governed by the MVT License 1.1 that can be found at +# https://license.mvt.re/1.1/ + +import logging +import sqlite3 +from typing import Optional + +from mvt.common.module import DatabaseNotFoundError +from mvt.common.module_types import ( + ModuleAtomicResult, + ModuleResults, + ModuleSerializedResult, +) +from mvt.common.utils import convert_mactime_to_iso + +from ..base import IOSExtraction + +WHATSAPP_CONTACTS_BACKUP_IDS = [ + # SHA-1 of "AppDomainGroup-group.net.whatsapp.WhatsApp.shared-ContactsV2.sqlite" + "b8548dc30aa1030df0ce18ef08b882cf7ab5212f", +] +WHATSAPP_CONTACTS_ROOT_PATHS = [ + "private/var/mobile/Containers/Shared/AppGroup/*/ContactsV2.sqlite", +] + +# WhatsApp's standard disappearing-messages timer values, in seconds. +DISAPPEARING_DURATION_LABELS = { + 86400: "24 hours", + 604800: "7 days", + 1209600: "14 days", + 2592000: "30 days", + 7776000: "90 days", +} + +# Output field -> candidate columns in ZWAADDRESSBOOKCONTACT, in order of +# preference. WhatsApp renames columns across versions, so the query is built +# from the columns actually present in the database. +COLUMN_CANDIDATES = { + "whatsapp_id": ["ZWHATSAPPID"], + "lid": ["ZLID"], + "phone_number": ["ZPHONENUMBER"], + "localized_phone_number": ["ZLOCALIZEDPHONENUMBER"], + "full_name": ["ZFULLNAME"], + "given_name": ["ZGIVENNAME"], + "last_name": ["ZLASTNAME"], + "user_name": ["ZUSERNAME"], + "business_name": ["ZBUSINESSNAME"], + "about_text": ["ZABOUTTEXT"], + "about_emoji": ["ZABOUTEMOJI"], + "notes": ["ZNOTES"], + "disappearing_mode_duration": ["ZDISAPPEARINGMODEDURATION"], + "disappearing_mode_timestamp": ["ZDISAPPEARINGMODETIMESTAMP"], + "about_timestamp": ["ZABOUTTIMESTAMP"], + "about_expiration_timestamp": ["ZABOUTEXPIRATIONTIMESTAMP"], + "last_updated": ["ZLASTUPDATED"], + "phone_status": ["ZPHONESTATUS", "ZPHONENUMBERSTATUS"], + "sync_policy": ["ZSYNCPOLICY"], +} + +STRING_FIELDS = [ + "whatsapp_id", + "lid", + "phone_number", + "localized_phone_number", + "full_name", + "given_name", + "last_name", + "user_name", + "business_name", + "about_text", + "about_emoji", + "notes", +] + +DATE_FIELDS = [ + "disappearing_mode_timestamp", + "about_timestamp", + "about_expiration_timestamp", + "last_updated", +] + + +def _decode_string(value) -> Optional[str]: + # CoreData stores string attributes as UTF-8 blobs in some WhatsApp + # versions, so values can arrive as either bytes or str. + if value is None: + return None + if isinstance(value, bytes): + return value.decode("utf-8", "replace") + return str(value) + + +def _label_duration(duration) -> str: + if not duration: + return "off" + return DISAPPEARING_DURATION_LABELS.get( + int(duration), f"{int(duration)} seconds" + ) + + +def _describe_contact(record: ModuleAtomicResult) -> str: + contact = ( + record.get("whatsapp_id") + or record.get("lid") + or record.get("phone_number") + or "unknown" + ) + full_name = record.get("full_name") + if full_name: + contact = f"{contact} ({full_name})" + return contact + + +class WhatsappContacts(IOSExtraction): + """This module extracts WhatsApp contact records and per-contact + disappearing-messages settings from ContactsV2.sqlite. + + ChatStorage.sqlite does not record the disappearing-messages state of 1:1 + chats: the authoritative timer is stored on each contact record in this + database, alongside the mapping between a contact's LID and phone number + identifiers. + """ + + def __init__( + self, + file_path: Optional[str] = None, + target_path: Optional[str] = None, + results_path: Optional[str] = None, + module_options: Optional[dict] = None, + log: logging.Logger = logging.getLogger(__name__), + results: Optional[ModuleResults] = None, + ) -> None: + super().__init__( + file_path=file_path, + target_path=target_path, + results_path=results_path, + module_options=module_options, + log=log, + results=results, + ) + + def serialize(self, record: ModuleAtomicResult) -> ModuleSerializedResult: + records = [] + contact = _describe_contact(record) + + if record.get("disappearing_mode_timestamp"): + records.append( + { + "timestamp": record["disappearing_mode_timestamp"], + "module": self.__class__.__name__, + "event": "disappearing_mode_set", + "data": ( + f"WhatsApp disappearing messages timer set to " + f"'{record.get('disappearing_mode_label')}' " + f"for {contact}" + ), + } + ) + + if record.get("about_timestamp"): + data = f"WhatsApp about text of {contact} changed" + about_text = record.get("about_text") + if about_text: + data += f' to "{about_text}"' + records.append( + { + "timestamp": record["about_timestamp"], + "module": self.__class__.__name__, + "event": "about_changed", + "data": data, + } + ) + + if record.get("about_expiration_timestamp"): + records.append( + { + "timestamp": record["about_expiration_timestamp"], + "module": self.__class__.__name__, + "event": "about_expiration", + "data": ( + f"WhatsApp about text of {contact} scheduled " + f"to expire" + ), + } + ) + + if record.get("last_updated"): + records.append( + { + "timestamp": record["last_updated"], + "module": self.__class__.__name__, + "event": "contact_last_updated", + "data": f"WhatsApp contact record for {contact} updated", + } + ) + + return records + + def run(self) -> None: + try: + self._find_ios_database( + backup_ids=WHATSAPP_CONTACTS_BACKUP_IDS, + root_paths=WHATSAPP_CONTACTS_ROOT_PATHS, + ) + except DatabaseNotFoundError: + self.log.warning( + "Unable to find the WhatsApp ContactsV2.sqlite database in " + "this backup or filesystem dump. WhatsApp disappearing " + "messages settings and contact records cannot be extracted. " + "This database is often missing from incremental backups." + ) + return + + self.log.info( + "Found WhatsApp contacts database at path: %s", self.file_path + ) + + assert self.file_path is not None + conn = self._open_sqlite_db(self.file_path) + cur = conn.cursor() + try: + try: + cur.execute("PRAGMA table_info(ZWAADDRESSBOOKCONTACT)") + available_columns = {row[1] for row in cur.fetchall()} + except sqlite3.DatabaseError as exc: + self.log.error( + "Unable to read the ZWAADDRESSBOOKCONTACT table schema: %s", + exc, + ) + return + + if not available_columns: + self.log.warning( + "The WhatsApp contacts database does not contain a " + "ZWAADDRESSBOOKCONTACT table" + ) + return + + selected = {} + for field, candidates in COLUMN_CANDIDATES.items(): + for candidate in candidates: + if candidate in available_columns: + selected[field] = candidate + break + + # A record with no duration column is "unknown", not "off": the + # timer state cannot be determined from this database version. + has_duration = "disappearing_mode_duration" in selected + if not has_duration: + self.log.warning( + "The ZDISAPPEARINGMODEDURATION column is not present in " + "this WhatsApp contacts database: disappearing messages " + "state is unknown" + ) + + columns = ["Z_PK"] + list(selected.values()) + cur.execute( + f"SELECT {', '.join(columns)} FROM ZWAADDRESSBOOKCONTACT;" + ) + fields = ["row_pk"] + list(selected.keys()) + + for row in cur: + record = dict(zip(fields, row)) + + for field in STRING_FIELDS: + if field in record: + record[field] = _decode_string(record[field]) + else: + record[field] = None + + for field in DATE_FIELDS: + if record.get(field) is not None: + record[field] = ( + convert_mactime_to_iso(record[field]) or None + ) + else: + record[field] = None + + duration = record.get("disappearing_mode_duration") + if has_duration: + record["disappearing_mode_is_on"] = bool(duration) + record["disappearing_mode_label"] = _label_duration( + duration + ) + else: + record["disappearing_mode_duration"] = None + record["disappearing_mode_is_on"] = None + record["disappearing_mode_label"] = None + + record.setdefault("phone_status", None) + record.setdefault("sync_policy", None) + + self.results.append(record) + finally: + cur.close() + conn.close() + + total_ephemeral = sum( + 1 for record in self.results if record["disappearing_mode_is_on"] + ) + self.log.info( + "Extracted a total of %d WhatsApp contacts (%d with disappearing " + "messages enabled)", + len(self.results), + total_ephemeral, + ) diff --git a/tests/artifacts/ios_backup/1f/1f5a521220a3ad80ebfdc196978df8e7a2e49dee b/tests/artifacts/ios_backup/1f/1f5a521220a3ad80ebfdc196978df8e7a2e49dee new file mode 100644 index 0000000..1d27832 Binary files /dev/null and b/tests/artifacts/ios_backup/1f/1f5a521220a3ad80ebfdc196978df8e7a2e49dee differ diff --git a/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d b/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d new file mode 100644 index 0000000..b8ae144 Binary files /dev/null and b/tests/artifacts/ios_backup/7c/7c7fba66680ef796b916b067077cc246adacf01d differ diff --git a/tests/artifacts/ios_backup/b8/b8548dc30aa1030df0ce18ef08b882cf7ab5212f b/tests/artifacts/ios_backup/b8/b8548dc30aa1030df0ce18ef08b882cf7ab5212f new file mode 100644 index 0000000..132059c Binary files /dev/null and b/tests/artifacts/ios_backup/b8/b8548dc30aa1030df0ce18ef08b882cf7ab5212f 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_interactionc.py b/tests/ios_backup/test_interactionc.py new file mode 100644 index 0000000..eec2656 --- /dev/null +++ b/tests/ios_backup/test_interactionc.py @@ -0,0 +1,105 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 The MVT Authors. +# Use of this software is governed by the MVT License 1.1 that can be found at +# https://license.mvt.re/1.1/ + +from mvt.common.module import run_module +from mvt.ios.modules.mixed.interactionc import InteractionC +from mvt.ios.modules.mixed.whatsapp_contacts import WhatsappContacts + +from ..utils import get_ios_backup_folder + + +class TestInteractionCModule: + def test_extraction_with_whatsapp_contacts(self): + contacts = WhatsappContacts(target_path=get_ios_backup_folder()) + run_module(contacts) + + m = InteractionC(target_path=get_ios_backup_folder()) + m.dependency_modules = {WhatsappContacts: contacts} + run_module(m) + + assert len(m.results) == 3 + + incoming = next( + r for r in m.results if r["sender_identifier"] == "100000000000001@lid" + ) + assert incoming["direction"] == "INCOMING" + assert incoming["sender_resolved_phone_number"] == "+14155550100" + assert incoming["sender_resolved_name"] == "Alice Example" + + outgoing = next( + r for r in m.results if r["direction"] == "OUTGOING" + ) + assert outgoing["recipient_identifier"] == "+14155550100" + assert outgoing["recipient_resolved_name"] == "Alice Example" + assert outgoing["domain_resolved_phone_number"] == "+14155550100" + assert outgoing["domain_resolved_name"] == "Alice Example" + + sms = next( + r for r in m.results if r["bundle_id"] == "com.apple.MobileSMS" + ) + assert sms.get("sender_resolved_name") is None + assert sms["sender_display_name"] == "Bob Example" + + events = [entry["data"] for entry in m.timeline] + assert ( + "[net.whatsapp.WhatsApp] INCOMING from " + "Alice Example (+14155550100) to local user" in events + ) + assert ( + "[net.whatsapp.WhatsApp] OUTGOING from local user to " + "Alice Example (+14155550100)" in events + ) + assert ( + "[com.apple.MobileSMS] INCOMING from " + "Bob Example (+14155550101) to local user" in events + ) + + # The creation date is only serialized when it diverges from the + # start date; the SMS record was created 90 days after the event. + creation_events = [ + entry + for entry in m.timeline + if entry["event"] == "interactions_creation_date" + ] + assert len(creation_events) == 1 + assert creation_events[0]["timestamp"] == "2025-12-09 12:26:40.000000" + assert creation_events[0]["data"] == ( + "Interaction record created 90 days after the event: " + "[com.apple.MobileSMS] INCOMING from " + "Bob Example (+14155550101) to local user" + ) + + # Per-contact aggregate dates use contact-centric data strings. + first_seen = [ + entry + for entry in m.timeline + if entry["event"] == "first_incoming_sender_date" + ] + assert len(first_seen) == 1 + assert first_seen[0]["timestamp"] == "2025-09-03 13:46:40.000000" + assert first_seen[0]["data"] == ( + "First incoming interaction from Bob Example (+14155550101)" + ) + assert ( + "Last incoming interaction from Bob Example (+14155550101)" + in events + ) + + def test_extraction_without_whatsapp_contacts(self): + # Without the WhatsappContacts dependency the module still runs, and + # unresolvable LIDs are shown as-is. + m = InteractionC(target_path=get_ios_backup_folder()) + run_module(m) + + assert len(m.results) == 3 + events = [entry["data"] for entry in m.timeline] + assert ( + "[net.whatsapp.WhatsApp] INCOMING from " + "100000000000001@lid to local user" in events + ) + assert ( + "[net.whatsapp.WhatsApp] OUTGOING from local user to " + "+14155550100" in events + ) diff --git a/tests/ios_backup/test_whatsapp.py b/tests/ios_backup/test_whatsapp.py index 864fb84..3bc920d 100644 --- a/tests/ios_backup/test_whatsapp.py +++ b/tests/ios_backup/test_whatsapp.py @@ -6,8 +6,79 @@ import logging from mvt.common.indicators import Indicators +from mvt.common.module import run_module from mvt.ios.modules.mixed.whatsapp import Whatsapp +from ..utils import get_ios_backup_folder + + +def test_extraction(): + m = Whatsapp(target_path=get_ios_backup_folder()) + run_module(m) + + 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"] == "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 + assert alice["stored_message_count"] == 2 + + group = next(s for s in sessions if s["partner_name"] == "Example Group") + assert group["group_creation_date"] == "2025-08-21 20:13:20.000000" + assert group["first_stored_message_date"] == "2025-08-29 22:40:00.000000" + # The last stored message predates the session's own last-message date: + # the newest message in this chat was deleted. + 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, 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)" + ) + assert events[("chat_last_message", "2025-08-28 18:53:20.000000")] == ( + "Last message in WhatsApp chat with " + "'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' " + "(120000000000000001@g.us) was created" + ) + assert ("chat_first_message", "2025-08-29 22:40:00.000000") in events + assert ("chat_last_message", "2025-08-31 02:26:40.000000") in events + + assert len(m.alertstore.alerts) == 0 + def test_collect_url_results_includes_expansion(): module = Whatsapp( diff --git a/tests/ios_backup/test_whatsapp_contacts.py b/tests/ios_backup/test_whatsapp_contacts.py new file mode 100644 index 0000000..a91b8f1 --- /dev/null +++ b/tests/ios_backup/test_whatsapp_contacts.py @@ -0,0 +1,83 @@ +# Mobile Verification Toolkit (MVT) +# Copyright (c) 2021-2026 The MVT Authors. +# Use of this software is governed by the MVT License 1.1 that can be found at +# https://license.mvt.re/1.1/ + +from mvt.common.module import run_module +from mvt.ios.modules.mixed.whatsapp_contacts import WhatsappContacts + +from ..utils import get_ios_backup_folder + + +class TestWhatsappContactsModule: + def test_extraction(self): + m = WhatsappContacts(target_path=get_ios_backup_folder()) + run_module(m) + assert len(m.results) == 2 + + alice = next(r for r in m.results if r["given_name"] == "Alice") + assert alice["full_name"] == "Alice Example" + assert alice["phone_number"] == "+14155550100" + assert alice["whatsapp_id"] == "14155550100@s.whatsapp.net" + assert alice["lid"] == "100000000000001@lid" + assert alice["user_name"] == "alice.example" + assert alice["disappearing_mode_duration"] == 86400.0 + assert alice["disappearing_mode_is_on"] is True + assert alice["disappearing_mode_label"] == "24 hours" + assert alice["disappearing_mode_timestamp"] == "2025-07-23 21:46:40.000000" + assert alice["about_timestamp"] == "2025-07-12 08:00:00.000000" + assert alice["about_expiration_timestamp"] == "2025-08-16 01:20:00.000000" + assert alice["last_updated"] == "2025-08-04 11:33:20.000000" + + bob = next(r for r in m.results if r["given_name"] == "Bob") + assert bob["lid"] is None + assert bob["disappearing_mode_duration"] is None + assert bob["disappearing_mode_is_on"] is False + assert bob["disappearing_mode_label"] == "off" + assert bob["disappearing_mode_timestamp"] is None + + # Alice: disappearing_mode_set, about_changed, about_expiration and + # contact_last_updated. Bob: contact_last_updated only. + assert len(m.timeline) == 5 + + events = { + (entry["event"], entry["timestamp"]): entry["data"] + for entry in m.timeline + } + assert ( + "24 hours" + in events[("disappearing_mode_set", "2025-07-23 21:46:40.000000")] + ) + assert ( + "14155550100@s.whatsapp.net (Alice Example)" + in events[("disappearing_mode_set", "2025-07-23 21:46:40.000000")] + ) + assert ( + 'changed to "Hey there! I am using WhatsApp."' + in events[("about_changed", "2025-07-12 08:00:00.000000")] + ) + assert ( + "scheduled to expire" + in events[("about_expiration", "2025-08-16 01:20:00.000000")] + ) + + updated = [ + entry["data"] + for entry in m.timeline + if entry["event"] == "contact_last_updated" + ] + assert len(updated) == 2 + assert all( + entry["timestamp"] == "2025-08-04 11:33:20.000000" + for entry in m.timeline + if entry["event"] == "contact_last_updated" + ) + assert any("14155550101@s.whatsapp.net (Bob Example)" in d for d in updated) + + assert len(m.alertstore.alerts) == 0 + + def test_missing_database(self, tmp_path): + m = WhatsappContacts(target_path=str(tmp_path)) + run_module(m) + assert m.results == [] + assert len(m.alertstore.alerts) == 0 diff --git a/tests/ios_fs/test_filesystem.py b/tests/ios_fs/test_filesystem.py index 9fa664f..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) == 15 - assert len(m.timeline) == 15 + 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) == 15 - assert len(m.timeline) == 15 + assert len(m.results) == 23 + assert len(m.timeline) == 23 assert len(m.alertstore.alerts) == 1