Merge branch 'main' into feature/installed-module-packages

2026-09-03 08:30:51 +02:00 · 2026-08-19 23:13:58 +02:00
parent 8eefcb706a dd8bd2cb01
commit 31f3937bfd
14 changed files with 1066 additions and 19 deletions
@@ -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.
+
@@ -31,7 +31,6 @@ nav:
    - Introduction: "introduction.md"
    - Installation: "install.md"
    - Command Completion: "command_completion.md"
-    - Custom CLI Commands: "custom_commands.md"
    - Using Docker: "docker.md"
    - MVT for iOS:
        - iOS Forensic Methodology: "ios/methodology.md"
@@ -50,5 +49,7 @@ nav:
        - Check an Android Backup (SMS messages): "android/backup.md"
        - Check Android Intrusion Logs: "android/intrusion_logs.md"
    - Indicators of Compromise: "iocs.md"
-    - Development: "development.md"
+    - Development:
+        - Development Instructions: "development.md"
+        - Custom CLI Commands: "custom_commands.md"
    - License: "license.md"
@@ -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,
@@ -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))
@@ -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
@@ -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,
+        )
@@ -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
+        )
@@ -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(
@@ -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
@@ -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